java - java中实现阻塞队列的问题

Question

我正在java中实现一个线程安全的阻塞队列，但看起来线程正在一个接一个地运行，而不是并行运行。谁能帮我找出我做错了什么？我的代码如下:

package com.example;

import java.util.LinkedList;
import java.util.List;

class Producer implements Runnable{

    BlockingQueue blockingQueue;

    public Producer(BlockingQueue blockingQueue) {
        this.blockingQueue = blockingQueue;
    }

    @Override
    public void run() {
        int counter = 0;
        while (true)
        {
            try
            {
                blockingQueue.enqueue(counter++);
            }
            catch (InterruptedException ex)
            {
                ex.printStackTrace();
            }
        }
    }
}

class Consumer implements Runnable{

    BlockingQueue blockingQueue;

    public Consumer(BlockingQueue blockingQueue) {
        this.blockingQueue = blockingQueue;
    }

    @Override
    public void run() {


        while (true)
        {
            try
            {
                blockingQueue.dequeue();
            }
            catch (InterruptedException ex)
            {
                ex.printStackTrace();
            }
        }
    }
}

public class Test{
    public static void main(String[] args) {

        BlockingQueue blockingQueue = new BlockingQueue(10);
        Thread producer = new Thread(new Producer(blockingQueue), "Prod");
        Thread consumer = new Thread(new Consumer(blockingQueue), "Cons");
        producer.start();
        consumer.start();
    }
}

class BlockingQueue {

    private List queue = new LinkedList();
    private int  limit = 10;

    public BlockingQueue(int limit){
        this.limit = limit;
    }


    public synchronized void enqueue(Object item)
            throws InterruptedException  {

        while(this.queue.size() == this.limit) {
            System.out.println("Wait Enque : "+Thread.currentThread().getName());
            wait();
        }

        Thread.sleep(1000);
        System.out.println("Add Item : " + Thread.currentThread().getName());
        this.queue.add(item);
        notifyAll();
    }


    public synchronized Object dequeue()
            throws InterruptedException{

        while(this.queue.size() == 0){
            System.out.println("Wait Denque : "+Thread.currentThread().getName());
            wait();
        }

        Thread.sleep(1000);
        System.out.println("Remove Item : " + Thread.currentThread().getName());
        notifyAll();
        return this.queue.remove(0);
    }

}

我是多线程新手。

这是我得到的输出:

Add Item : Prod
Add Item : Prod
Add Item : Prod
Add Item : Prod
Add Item : Prod
Add Item : Prod
Add Item : Prod
Add Item : Prod
Add Item : Prod
Add Item : Prod
Wait Enque : Prod
Remove Item : Cons
Remove Item : Cons
Remove Item : Cons
Remove Item : Cons
Remove Item : Cons
Remove Item : Cons
Remove Item : Cons
Remove Item : Cons
Remove Item : Cons
Remove Item : Cons
Wait Denque : Cons
Add Item : Prod
Add Item : Prod
Add Item : Prod

Answer 1

您的 BlockingQueue 实现演示了应该避免的“工作时 sleep ”模式。

出于某种原因，您决定在数据结构的同步方法入队和出队中调用Thread.sleep。我认为根本没有必要。这些方法至少应该做的是以线程安全的方式使用共享的可变状态(即您的队列)。除了这些方法中对 Thread.sleep() 的调用之外的所有内容都是一个很好的首次尝试:

public synchronized void enqueue(Object item)
        throws InterruptedException {

    while (this.queue.size() == this.limit) {
        System.out.println("Wait Enque : "+Thread.currentThread().getName());
        wait();
    }

    System.out.println("Add Item : " + item.toString() + " " + Thread.currentThread().getName());
    this.queue.add(item);
    notifyAll();
}

与出队类似。

然而，你的线程正在做的是它们非常贪婪:-)。也许您实际上应该对线程的 run 方法内的出列项执行一些操作:

    while (true)
    {
        try
        {
            Object deq = blockingQueue.dequeue();
            Thread.sleep(1000); // sleeping to simulate using the de-queued item

        }
        catch (InterruptedException ex)
        {
            ex.printStackTrace();
        }
    }

所以，实际上，我所做的就是从数据结构的方法中取出 sleep 部分。我得到以下交错输出以及我期望的输出:

Add Item : 0 Prod
Remove Item : 0 Cons
Add Item : 1 Prod
Remove Item : 1 Cons
Add Item : 2 Prod
Add Item : 3 Prod
Remove Item : 2 Cons
Add Item : 4 Prod
Add Item : 5 Prod

当然，我们可以提出一些改进建议:

1. 将所有字段设为最终。
2. 使用 CountDownLatch 之类的东西来确保一定的公平性。

看看java.util.concurrent.LinkedBlockingQueue也可以获得一些见解!