如何本身實現一個BlockingQueue

一、synchronized模式

線程請求數據

@AllArgsConstructor
@Getter
public class Request {
    private final String name;

    @Override
    public String toString() {
        return "[ Request " + name + " ]";
    }
}

線程隊列

public class RequestQueue {
    private final Queue<Request> queue = new LinkedList<>();
    public synchronized Request getRequest() {
        while (queue.peek() == null) {
            try {
                wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        return queue.remove();
    }
    public synchronized void putRequest(Request request) {
        queue.offer(request);
        notifyAll();
    }
}

數據插入線程

@AllArgsConstructor
public class ClientThread implements Runnable {
    private final Random random;
    private final RequestQueue requestQueue;
    @Override
    public void run() {
        for (int i = 0;i < 10000;i++) {
            Request request = new Request("No." + i);
            System.out.println(Thread.currentThread().getName() + " Requests " + request);
            requestQueue.putRequest(request);
            try {
                Thread.sleep(random.nextInt(1000));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

數據取出線程

@AllArgsConstructor
public class ServerThread implements Runnable {
    private final Random random;
    private final RequestQueue requestQueue;
    @Override
    public void run() {
        for (int i = 0;i < 10000;i++) {
            Request request = requestQueue.getRequest();
            System.out.println(Thread.currentThread().getName() + " handles " + request);
            try {
                Thread.sleep(random.nextInt(1000));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

主方法

public class Main {
    public static void main(String[] args) {
        RequestQueue requestQueue = new RequestQueue();
        new Thread(new ClientThread(new Random(3141592L),requestQueue)).start();
        new Thread(new ServerThread(new Random(6535897L),requestQueue)).start();
    }
}

運行結果：

Thread-0 Requests [ Request No.0 ]
Thread-1 handles [ Request No.0 ]
Thread-0 Requests [ Request No.1 ]
Thread-0 Requests [ Request No.2 ]
Thread-1 handles [ Request No.1 ]
Thread-1 handles [ Request No.2 ]
Thread-0 Requests [ Request No.3 ]
Thread-1 handles [ Request No.3 ]
Thread-0 Requests [ Request No.4 ]
Thread-1 handles [ Request No.4 ]
Thread-0 Requests [ Request No.5 ]
Thread-0 Requests [ Request No.6 ]
Thread-1 handles [ Request No.5 ]

......

二、lock模式

惟一不一樣的是線程隊列，其餘同synchronized模式

public class RequestQueue {
    private final Queue<Request> queue = new LinkedList<>();
    private Lock lock = new ReentrantLock();
    private Condition condition = lock.newCondition();
    public  Request getRequest() {
        try {
            lock.lock();
            while (queue.peek() == null) {
                try {
                    condition.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            return queue.remove();
        } finally {
            lock.unlock();
        }
    }
    public void putRequest(Request request) {
        try {
            lock.lock();
            queue.offer(request);
            condition.signalAll();
        } finally {
            lock.unlock();
        }
    }
}