LinkedBlockingQueue的offer与put的区别

向右看齐 2022-09-24 01:28 190阅读 0赞

首先，看一下LinkedBlockingQueue的put方法的源码：

/**
         * Inserts the specified element at the tail of this queue, waiting if
         * necessary for space to become available.
         *
         * @throws InterruptedException {@inheritDoc}
         * @throws NullPointerException {@inheritDoc}
         */
        public void put(E e) throws InterruptedException {
            if (e == null) throw new NullPointerException();
            // Note: convention in all put/take/etc is to preset local var
            // holding count negative to indicate failure unless set.
            int c = -1;
            Node<E> node = new Node<E>(e);
            final ReentrantLock putLock = this.putLock;
            final AtomicInteger count = this.count;
            putLock.lockInterruptibly();
            try {
                /*
                 * Note that count is used in wait guard even though it is
                 * not protected by lock. This works because count can
                 * only decrease at this point (all other puts are shut
                 * out by lock), and we (or some other waiting put) are
                 * signalled if it ever changes from capacity. Similarly
                 * for all other uses of count in other wait guards.
                 */
                while (count.get() == capacity) {
                    notFull.await();
                }
                enqueue(node);
                c = count.getAndIncrement();
                if (c + 1 < capacity)
                    notFull.signal();
            } finally {
                putLock.unlock();
            }
            if (c == 0)
                signalNotEmpty();
        }

注释的意思是：往队尾插入指定的元素，如果空间不能用则挂起，等待唤醒。

offer方法的源码：

/**
         * Inserts the specified element at the tail of this queue if it is
         * possible to do so immediately without exceeding the queue's capacity,
         * returning {@code true} upon success and {@code false} if this queue
         * is full.
         * When using a capacity-restricted queue, this method is generally
         * preferable to method {@link BlockingQueue#add add}, which can fail to
         * insert an element only by throwing an exception.
         *
         * @throws NullPointerException if the specified element is null
         */
        public boolean offer(E e) {
            if (e == null) throw new NullPointerException();
            final AtomicInteger count = this.count;
            if (count.get() == capacity)
                return false;
            int c = -1;
            Node<E> node = new Node<E>(e);
            final ReentrantLock putLock = this.putLock;
            putLock.lock();
            try {
                if (count.get() < capacity) {
                    enqueue(node);
                    c = count.getAndIncrement();
                    if (c + 1 < capacity)
                        notFull.signal();
                }
            } finally {
                putLock.unlock();
            }
            if (c == 0)
                signalNotEmpty();
            return c >= 0;
        }

注释的意思是：往队尾插入指定的元素，如果空间不能用则忽略此次请求。那意思是不是有可能丢失元素呢？

看下边的例子：

import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.LinkedBlockingQueue;
    import java.util.concurrent.TimeUnit;
    
    public class BlockingQueueTest {
    	public static void main(String[] args) {
    		ExecutorService executor = Executors.newCachedThreadPool();
    		Basket basket = new Basket();
    		executor.execute(new Producer("生产者A", basket));
    		executor.execute(new Producer("生产者B", basket));
    		executor.execute(new Consumer("消费者C", basket));
    		
    		
    	      // 程序运行5s后，所有任务停止
          try {
              Thread.sleep(1000 * 3);
          } catch (InterruptedException e) {
          }
          executor.shutdownNow();
    	}
    }
    
    class Basket{
    	BlockingQueue<String> basket = new LinkedBlockingQueue<>(3);
    	private static int i = 0;
    	
    	public void produce() throws InterruptedException {
    		basket.put("An apple" + i++);
    //		basket.offer("An apple" + i++);
    	}
    	
    	public String consume() throws InterruptedException {
    		return basket.take();
    	}
    }
    class Producer implements Runnable{
    	 private String role;
    	 private Basket basket;
    	 
    	 public Producer(String role, Basket basket) {
    		 this.role = role;
            this.basket = basket;
    	 }
    	@Override
    	public void run() {
    			try {
    				while (true) {
    					System.out.println("生产者,准备生产苹果：" + role);
                        basket.produce();
                        System.out.println("生产者,生产苹果完毕：" + role);
    					TimeUnit.MILLISECONDS.sleep(300);
    				}
    			} catch (InterruptedException e) {
    			}
    	}
    }
    class Consumer implements Runnable{
    	private String role;
    	 private Basket basket;
    	 
    	 public Consumer(String role, Basket basket) {
    		 this.role = role;
             this.basket = basket;
    	 }
    	 
    	@Override
    	public void run() {
    		try {
    			while (true) {
    				// 消费苹果
                    System.out.println("消费者,准备消费苹果：" + role);
                    System.out.println(basket.consume());
                    System.out.println("消费者,消费苹果完毕：" + role);
    				TimeUnit.MILLISECONDS.sleep(300);
    			}
    		} catch (InterruptedException e) {
    		}
    	}
    }

结果如下：

生产者,准备生产苹果：生产者B  
生产者,准备生产苹果：生产者A  
生产者,生产苹果完毕：生产者B  
生产者,生产苹果完毕：生产者A  
消费者,准备消费苹果：消费者C  
An apple0  
消费者,消费苹果完毕：消费者C  
生产者,准备生产苹果：生产者B  
生产者,准备生产苹果：生产者A  
生产者,生产苹果完毕：生产者A  
消费者,准备消费苹果：消费者C  
An apple1  
消费者,消费苹果完毕：消费者C  
生产者,生产苹果完毕：生产者B  
生产者,准备生产苹果：生产者A  
生产者,生产苹果完毕：生产者A  
生产者,准备生产苹果：生产者B  
消费者,准备消费苹果：消费者C  
An apple2  
消费者,消费苹果完毕：消费者C  
生产者,生产苹果完毕：生产者B  
生产者,准备生产苹果：生产者A  
生产者,准备生产苹果：生产者B  
消费者,准备消费苹果：消费者C  
An apple3  
消费者,消费苹果完毕：消费者C  
生产者,生产苹果完毕：生产者A  
生产者,准备生产苹果：生产者A  
消费者,准备消费苹果：消费者C  
An apple4  
消费者,消费苹果完毕：消费者C  
生产者,生产苹果完毕：生产者B  
生产者,准备生产苹果：生产者B  
消费者,准备消费苹果：消费者C  
An apple5  
消费者,消费苹果完毕：消费者C  
生产者,生产苹果完毕：生产者A  
生产者,准备生产苹果：生产者A  
消费者,准备消费苹果：消费者C  
An apple6  
消费者,消费苹果完毕：消费者C  
生产者,生产苹果完毕：生产者B  
消费者,准备消费苹果：消费者C  
生产者,准备生产苹果：生产者B  
An apple7  
消费者,消费苹果完毕：消费者C  
生产者,生产苹果完毕：生产者A  
消费者,准备消费苹果：消费者C  
生产者,准备生产苹果：生产者A  
An apple8  
消费者,消费苹果完毕：消费者C  
生产者,生产苹果完毕：生产者A  
生产者,准备生产苹果：生产者A  
消费者,准备消费苹果：消费者C  
An apple9  
消费者,消费苹果完毕：消费者C  
生产者,生产苹果完毕：生产者B

然后，我们把这个地方改一下：

public void produce() throws InterruptedException {
    //		basket.put("An apple" + i++);
    		basket.offer("An apple" + i++);
    	}

再看结果：

生产者,准备生产苹果：生产者B  
生产者,准备生产苹果：生产者A  
生产者,生产苹果完毕：生产者A  
生产者,生产苹果完毕：生产者B  
消费者,准备消费苹果：消费者C  
An apple0  
消费者,消费苹果完毕：消费者C  
生产者,准备生产苹果：生产者A  
生产者,准备生产苹果：生产者B  
生产者,生产苹果完毕：生产者A  
生产者,生产苹果完毕：生产者B  
消费者,准备消费苹果：消费者C  
An apple1  
消费者,消费苹果完毕：消费者C  
生产者,准备生产苹果：生产者B  
生产者,准备生产苹果：生产者A  
生产者,生产苹果完毕：生产者B  
生产者,生产苹果完毕：生产者A  
消费者,准备消费苹果：消费者C  
An apple2  
消费者,消费苹果完毕：消费者C  
生产者,准备生产苹果：生产者B  
生产者,准备生产苹果：生产者A  
生产者,生产苹果完毕：生产者B  
生产者,生产苹果完毕：生产者A  
消费者,准备消费苹果：消费者C  
An apple3  
消费者,消费苹果完毕：消费者C  
生产者,准备生产苹果：生产者A  
生产者,准备生产苹果：生产者B  
生产者,生产苹果完毕：生产者B  
生产者,生产苹果完毕：生产者A  
消费者,准备消费苹果：消费者C  
An apple4  
消费者,消费苹果完毕：消费者C  
生产者,准备生产苹果：生产者B  
生产者,生产苹果完毕：生产者B  
生产者,准备生产苹果：生产者A  
生产者,生产苹果完毕：生产者A  
消费者,准备消费苹果：消费者C  
An apple6  
消费者,消费苹果完毕：消费者C  
生产者,准备生产苹果：生产者A  
生产者,准备生产苹果：生产者B  
生产者,生产苹果完毕：生产者A  
生产者,生产苹果完毕：生产者B  
消费者,准备消费苹果：消费者C  
An apple8  
消费者,消费苹果完毕：消费者C  
生产者,准备生产苹果：生产者B  
生产者,准备生产苹果：生产者A  
生产者,生产苹果完毕：生产者B  
生产者,生产苹果完毕：生产者A  
消费者,准备消费苹果：消费者C  
An apple10  
消费者,消费苹果完毕：消费者C  
生产者,准备生产苹果：生产者B  
生产者,生产苹果完毕：生产者B  
生产者,准备生产苹果：生产者A  
生产者,生产苹果完毕：生产者A  
消费者,准备消费苹果：消费者C  
An apple12  
消费者,消费苹果完毕：消费者C  
生产者,准备生产苹果：生产者B  
生产者,准备生产苹果：生产者A  
生产者,生产苹果完毕：生产者B  
生产者,生产苹果完毕：生产者A  
消费者,准备消费苹果：消费者C  
An apple14  
消费者,消费苹果完毕：消费者C

对比一下两次结果：你会发现，怎么第二次苹果编号不是连续的。是的，那就是我们上边阐述的问题，offer方法不会阻塞，如果不能插入队列直接返回，有可能造成数据丢失。而，put方法会阻塞，等待消费完有空位时，唤醒put线程。

相同的道理：获取元素，take方法是阻塞的，poll方法不是阻塞的。