一.使用

  1. public class Test {
  3.     public static void main(String[] args) throws InterruptedException {
  4.         SynchronousQueue<Integer> queue = new SynchronousQueue<>();
  5.         Producer p1 = new Producer("p1", queue, 1);
  6.         Producer p2 = new Producer("p2", queue, 2);
  8.         Consumer c1 = new Consumer("c1", queue);
  9.         Consumer c2 = new Consumer("c2", queue);
  11.         c1.start();
  12.         Thread.sleep(1000);
  13.         c2.start();
  14.         Thread.sleep(1000);
  15.         p1.start();
  16.         Thread.sleep(1000);
  17.         p2.start();
  18.     }
  21. }
  23. class Producer extends Thread {
  24.     private SynchronousQueue<Integer> queue;
  25.     private int n;
  27.     public Producer(String name, SynchronousQueue<Integer> queue, int n) {
  28.         super(name);
  29.         this.queue = queue;
  30.         this.n = n;
  31.     }
  33.     @Override
  34.     public void run() {
  35.         System.out.println(getName() + " offer " + queue.offer(n));
  36.     }
  37. }
  39. class Consumer extends Thread {
  40.     private SynchronousQueue<Integer> queue;
  42.     public Consumer(String name, SynchronousQueue<Integer> queue) {
  43.         super(name);
  44.         this.queue = queue;
  45.     }
  47.     @SneakyThrows
  48.     @Override
  49.     public void run() {
  50.         System.out.println(getName() + " take " + queue.take());
  51.     }
  52. }
  54. 输出：
  55. p1 offer true
  56. c2 take 1
  57. p2 offer true
  58. c1 take 2

二.源码

SynchronousQueue构造方法

  1. public SynchronousQueue() {
  2.     this(false);
  3. }
  5. public SynchronousQueue(boolean fair) {
  6.     transferer = fair ? new TransferQueue<E>() : new TransferStack<E>();
  7. }

take

  1. public E take() throws InterruptedException {
  2.     E e = transferer.transfer(null, false, 0);
  3.     if (e != null)
  4.         return e;
  5. }
  7. E transfer(E e, boolean timed, long nanos) {
  8.     SNode s = null; 
  9.     // REQUEST:消费，DATA:生产
  10.     int mode = (e == null) ? REQUEST : DATA;
  11.     for (;;) {
  12.         SNode h = head;
  13.         if (h == null || h.mode == mode) {
  14.             // 入栈
  15.             if (casHead(h, s = snode(s, e, h, mode))) {
  16.                 SNode m = awaitFulfill(s, timed, nanos);
  17.                 return (E) ((mode == REQUEST) ? m.item : s.item);
  18.             }
  19.         }
  20.     }
  21. }
  23. SNode awaitFulfill(SNode s, boolean timed, long nanos) {
  24.     Thread w = Thread.currentThread();
  25.     int spins = (shouldSpin(s) ?
  26.                  (timed ? maxTimedSpins : maxUntimedSpins) : 0);
  27.     for (;;) {
  28.         SNode m = s.match;
  29.         if (m != null)
  30.             // 4.唤醒返回
  31.             return m;
  32.         if (spins > 0)
  33.             // 1.自旋
  34.             spins = shouldSpin(s) ? (spins-1) : 0;
  35.         else if (s.waiter == null)
  36.             // 2.设置线程
  37.             s.waiter = w;
  38.         else if (!timed)
  39.             // 3.阻塞
  40.             LockSupport.park(this);
  41.         else if (nanos > spinForTimeoutThreshold)
  42.             LockSupport.parkNanos(this, nanos);
  43.     }
  44. }

offer

  1. public boolean offer(E e) {
  2.     if (e == null) throw new NullPointerException();
  3.     return transferer.transfer(e, true, 0) != null;
  4. }
  6. E transfer(E e, boolean timed, long nanos) {
  7.     SNode s = null; 
  8.     // REQUEST:消费，DATA:生产
  9.     int mode = (e == null) ? REQUEST : DATA;
  11.     for (;;) {
  12.         SNode h = head;
  13.         if (!isFulfilling(h.mode)) {
  14.             // 入栈
  15.             if (casHead(h, s=snode(s, e, h, FULFILLING|mode))) {
  16.                 for (;;) { 
  17.                     SNode m = s.next;      
  18.                     SNode mn = m.next;
  19.                     if (m.tryMatch(s)) {
  20.                         // 设置head
  21.                         casHead(s, mn);    
  22.                         return (E) ((mode == REQUEST) ? m.item : s.item);
  23.                     } 
  24.                 }
  25.             }
  26.         }
  27.     }
  28. }
  30. boolean tryMatch(SNode s) {
  31.     // this对象的match属性赋值
  32.     if (match == null &&
  33.         UNSAFE.compareAndSwapObject(this, matchOffset, null, s)) {
  34.         Thread w = waiter;
  35.         if (w != null) {
  36.             waiter = null;
  37.             // 唤醒
  38.             LockSupport.unpark(w);
  39.         }
  40.         return true;
  41.     }
  42.     return match == s;
  43. }