SynchronousQueue和LinkedBlockingQueue区别

在多线程开发中，会用到SynchronousQueue(new CachedThreadPool())和LinkedBlockingQueue(new FixedThreadPoll())

我们来简单分析一下这两个队列的区别

SynchronousQueue:

　　offer()：当线程offer操作时，当same mode时，加入队列失败，即时返回 （如果是put操作，元素会存储到队列中，并且阻塞等待）；

当complimentary mode时，立即把元素transfer给等待的take线程

        take()：线程take操作，当same mode时，该线程把元素存储到队列中，并且阻塞等待（如果是poll操作，元素会加入队列失败，即时返回）；

当complimentary mode时，立即在队列中找到等待的put线程关联的元素，取出来，返回

LinkedBlockingQueue

　　offer(): 线程把元素放入队列中（多线程并发竞争），返回，超过bound，返回失败

/** Lock held by take, poll, etc */
    private final ReentrantLock takeLock = new ReentrantLock();

    /** Wait queue for waiting takes */
    private final Condition notEmpty = takeLock.newCondition();

    /** Lock held by put, offer, etc */
    private final ReentrantLock putLock = new ReentrantLock();

    /** Wait queue for waiting puts */
    private final Condition notFull = putLock.newCondition();


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;
    }

       take(); 队列不为空时，获取元素（多线程并发竞争），为空时，阻塞等待

public E take() throws InterruptedException {
        E x;
        int c = -1;
        final AtomicInteger count = this.count;
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lockInterruptibly();
        try {
            while (count.get() == 0) {
                notEmpty.await();
            }
            x = dequeue();
            c = count.getAndDecrement();
            if (c > 1)
                notEmpty.signal();
        } finally {
            takeLock.unlock();
        }
        if (c == capacity)
            signalNotFull();
        return x;
    }

设想一种场景：

　　当有大量线程在offer和take时，

　　1、LinkedBlockingQueue在入队和出队时，并发竞争激烈，cpu线程切换频繁，性能较低；

 　　2、SynchronousQueue中每一个元素对应一个put线程、一个take线程，不会存在锁竞争

　　但是反过来 SynchronousQueue需要的线程数较多，如果take消费不及时，会导致put线程阻塞（如果是使用offer的话，会加入队列失败），

在new CachedThreadPool()中，当执行execute()时

public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        /*
         * Proceed in 3 steps:
         *
         * 1. If fewer than corePoolSize threads are running, try to
         * start a new thread with the given command as its first
         * task.  The call to addWorker atomically checks runState and
         * workerCount, and so prevents false alarms that would add
         * threads when it shouldn't, by returning false.
         *
         * 2. If a task can be successfully queued, then we still need
         * to double-check whether we should have added a thread
         * (because existing ones died since last checking) or that
         * the pool shut down since entry into this method. So we
         * recheck state and if necessary roll back the enqueuing if
         * stopped, or start a new thread if there are none.
         *
         * 3. If we cannot queue task, then we try to add a new
         * thread.  If it fails, we know we are shut down or saturated
         * and so reject the task.
         */
        int c = ctl.get();
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
　　　　　　　//same mode时，入队失败，返回false
            //complimentary mode时 入队成功返回true
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            if (! isRunning(recheck) && remove(command))
                reject(command);
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
　　　　　　 //当大批量任务进来，command执行时间较长，会大量入队失败
　　　　　　 //，并且cachedThreadPool，最大线程数为MAX，会无节制的创建大量线程
        else if (!addWorker(command, false))
            reject(command);
    }