Java线程同步队列SynchronousQueue

╰+攻爆jí腚メ 2024-01-20 07:20 57阅读 0赞

## Java线程同步队列SynchronousQueue ##

SynchronousQueue是一个没有数据缓冲的阻塞队列。生产者线程插入数据(put)必须等待消费者的移除数据(take)，反之亦然。  
SynchronousQueue内部没有维护数据队列，故，开发者不能使用peek方法取数据。也不能像往常那样遍历队列。  
SynchronousQueue特别适用于数据传递和交换的场景。比如对于经典生产者-消费者线程模型：生产者和消费者之间等待对方的到来，然后交换数据，最后离开。

SynchronousQueue关键函数：

--------------------

iterator() 永远为空，里面没数据。

peek() 永远返回null。

put() 在队列里面放入一个数据元素，然后就一直等待下去，知道其他线程把这个数据元素取走。

offer() 在queue里放一个数据元素立即返回，如果刚好此时该数据元素被另一个线程把该数据取走，offer返回true，否则返回false。

take() 取出并且删除队列里的数据元素，如果取不到数据元素他会一直等。

poll() 取出并且删除队列里的数据元素，只有到碰巧另外一个线程正在往queue里offer数据或者put数据的时候，该方法才会取到东西。否则立即返回null。

isEmpty()永远返回true。

remainingCapacity() 永远为0。

remove()和removeAll() 永远为false。

--------------------

一个数据元素不会在SynchronousQueue 队列里面长时间停留，一旦有了数据存线程和数据取线程，数据元素很快就从存线程交换给取线程。该过程更像是一种信道或管道模型，数据通过SynchronousQueue队列从一端快速传递到另一端。显然这是一种高效的数据元素的交换方式。

private int seq = 0;
        private SynchronousQueue mQueue = new SynchronousQueue();
        
        private void test(){
            t2();
            t1();
        }
    
        private void t1() {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    while (true) {
                        try {
                            TimeUnit.SECONDS.sleep(3);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
    
                        try {
                            System.out.println("放入数据:" + seq);
                            mQueue.put(seq++);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }).start();
        }
    
        private void t2() {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    while (true) {
                        try {
                            System.out.println("从队列取数据...");
                            int data = (int) mQueue.take();
                            System.out.println("从队列取出数据:" + data);
    
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }).start();
        }

输出：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly96aGFuZ3BoaWwuYmxvZy5jc2RuLm5ldA_size_16_color_FFFFFF_t_70][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly96aGFuZ3BoaWwuYmxvZy5jc2RuLm5ldA_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/20/7a47bb91eac140df91cc9c6c09cb7c34.png