併發編程-隊列-SynchronousQueue

介紹

Java 6的併發編程包中的SynchronousQueue是一個沒有數據緩衝的BlockingQueue，生產者線程對其的插入操做put必須等待消費者的移除操做take，反過來也同樣。

不像ArrayBlockingQueue或LinkedListBlockingQueue，SynchronousQueue內部並無數據緩存空間，你不能調用peek()方法來看隊列中是否有數據元素，由於數據元素只有當你試着取走的時候纔可能存在，不取走而只想偷窺一下是不行的，固然遍歷這個隊列的操做也是不容許的。隊列頭元素是第一個排隊要插入數據的線程，而不是要交換的數據。數據是在配對的生產者和消費者線程之間直接傳遞的，並不會將數據緩衝數據到隊列中。能夠這樣來理解：生產者和消費者互相等待對方，握手，而後一塊兒離開。

SynchronousQueue的一個使用場景是在線程池裏。Executors.newCachedThreadPool()就使用了SynchronousQueue，這個線程池根據須要（新任務到來時）建立新的線程，若是有空閒線程則會重複使用，線程空閒了60秒後會被回收。

實現原理

阻塞隊列的實現方法有許多：

阻塞算法實現

阻塞算法實現一般在內部採用一個鎖來保證多個線程中的put()和take()方法是串行執行的。採用鎖的開銷是比較大的，還會存在一種狀況是線程A持有線程B須要的鎖，B必須一直等待A釋放鎖，即便A可能一段時間內由於B的優先級比較高而得不到時間片運行。因此在高性能的應用中咱們經常但願規避鎖的使用。

01	public class NativeSynchronousQueue<E> {

02	boolean putting = false;

03	E item = null;

04

05	public synchronized E take() throws InterruptedException {

06	while (item == null)

07

wait();

08	E e = item;

09	item = null;

10	notifyAll();

11

return e;

12

}

13

14	public synchronized void put(E e) throws InterruptedException {

15	if (e==null) return;

16	while (putting)

17

wait();

18	putting = true;

19

item = e;

20	notifyAll();

21	while (item!=null)

22

wait();

23	putting = false;

24	notifyAll();

25

}

26

}

信號量實現

經典同步隊列實現採用了三個信號量，代碼很簡單，比較容易理解：

01	public class SemaphoreSynchronousQueue<E> {

02	E item = null;

03	Semaphore sync = new Semaphore(0);

04	Semaphore send = new Semaphore(1);

05	Semaphore recv = new Semaphore(0);

06

07	public E take() throws InterruptedException {

08	recv.acquire();

09	E x = item;

10	sync.release();

11	send.release();

12

return x;

13

}

14

15	public void put (E x) throws InterruptedException{

16	send.acquire();

17

item = x;

18	recv.release();

19	sync.acquire();

20

}

21

}

在多核機器上，上面方法的同步代價仍然較高，操做系統調度器須要上千個時間片來阻塞或喚醒線程，而上面的實現即便在生產者put()時已經有一個消費者在等待的狀況下，阻塞和喚醒的調用仍然須要。

Java 5實現

01	public class Java5SynchronousQueue<E> {

02	ReentrantLock qlock = new ReentrantLock();

03	Queue waitingProducers = new Queue();

04	Queue waitingConsumers = new Queue();

05

06	static class Node extends AbstractQueuedSynchronizer {

07

E item;

08	Node next;

09

10	Node(Object x) { item = x; }

11	void waitForTake() { /* (uses AQS) */ }

12	E waitForPut() { /* (uses AQS) */ }

13

}

14

15	public E take() {

16	Node node;

17	boolean mustWait;

18	qlock.lock();

19	node = waitingProducers.pop();

20	if(mustWait = (node == null))

21	node = waitingConsumers.push(null);

22	qlock.unlock();

23

24	if (mustWait)

25	return node.waitForPut();

26

else

27	return node.item;

28

}

29

30	public void put(E e) {

31	Node node;

32	boolean mustWait;

33	qlock.lock();

34	node = waitingConsumers.pop();

35	if (mustWait = (node == null))

36	node = waitingProducers.push(e);

37	qlock.unlock();

38

39	if (mustWait)

40	node.waitForTake();

41

else

42	node.item = e;

43

}

44

}

Java 5的實現相對來講作了一些優化，只使用了一個鎖，使用隊列代替信號量也能夠容許發佈者直接發佈數據，而不是要首先從阻塞在信號量處被喚醒。

Java6實現

Java 6的SynchronousQueue的實現採用了一種性能更好的無鎖算法 — 擴展的「Dual stack and Dual queue」算法。性能比Java5的實現有較大提高。競爭機制支持公平和非公平兩種：非公平競爭模式使用的數據結構是後進先出棧(Lifo Stack)；公平競爭模式則使用先進先出隊列（Fifo Queue），性能上二者是至關的，通常狀況下，Fifo一般能夠支持更大的吞吐量，但Lifo能夠更大程度的保持線程的本地化。

代碼實現裏的Dual Queue或Stack內部是用鏈表(LinkedList)來實現的，其節點狀態爲如下三種狀況：

1. 持有數據 – put()方法的元素
2. 持有請求 – take()方法
3. 空

這個算法的特色就是任何操做均可以根據節點的狀態判斷執行，而不須要用到鎖。

其核心接口是Transfer，生產者的put或消費者的take都使用這個接口，根據第一個參數來區別是入列（棧）仍是出列（棧）。

01

/**

02	* Shared internal API for dual stacks and queues.

03

*/

04	static abstract class Transferer {

05

/**

06	* Performs a put or take.

07

*

08	* @param e if non-null, the item to be handed to a consumer;

09	*          if null, requests that transfer return an item

10	*          offered by producer.

11	* @param timed if this operation should timeout

12	* @param nanos the timeout, in nanoseconds

13	* @return if non-null, the item provided or received; if null,

14	*         the operation failed due to timeout or interrupt --

15	*         the caller can distinguish which of these occurred

16	*         by checking Thread.interrupted.

17

*/

18	abstract Object transfer(Object e, boolean timed, long nanos);

19

}

TransferQueue實現以下(摘自Java 6源代碼)，入列和出列都基於Spin和CAS方法：

view source

print?

01

/**

02	* Puts or takes an item.

03

*/

04	Object transfer(Object e, boolean timed, long nanos) {

05	/* Basic algorithm is to loop trying to take either of

06	* two actions:

07

*

08	* 1. If queue apparently empty or holding same-mode nodes,

09	*    try to add node to queue of waiters, wait to be

10	*    fulfilled (or cancelled) and return matching item.

11

*

12	* 2. If queue apparently contains waiting items, and this

13	*    call is of complementary mode, try to fulfill by CAS'ing

14	*    item field of waiting node and dequeuing it, and then

15	*    returning matching item.

16

*

17	* In each case, along the way, check for and try to help

18	* advance head and tail on behalf of other stalled/slow

19	* threads.

20

*

21	* The loop starts off with a null check guarding against

22	* seeing uninitialized head or tail values. This never

23	* happens in current SynchronousQueue, but could if

24	* callers held non-volatile/final ref to the

25	* transferer. The check is here anyway because it places

26	* null checks at top of loop, which is usually faster

27	* than having them implicitly interspersed.

28

*/

29

30	QNode s = null; // constructed/reused as needed

31	boolean isData = (e != null);

32

33	for (;;) {

34	QNode t = tail;

35	QNode h = head;

36	if (t == null || h == null)         // saw uninitialized value

37	continue;                       // spin

38

39	if (h == t || t.isData == isData) { // empty or same-mode

40	QNode tn = t.next;

41	if (t != tail)                  // inconsistent read

42

continue;

43	if (tn != null) {               // lagging tail

44	advanceTail(t, tn);

45

continue;

46

}

47	if (timed && nanos <= 0)        // can't wait

48	return null;

49	if (s == null)

50	s = new QNode(e, isData);

51	if (!t.casNext(null, s))        // failed to link in

52

continue;

53

54	advanceTail(t, s);              // swing tail and wait

55	Object x = awaitFulfill(s, e, timed, nanos);

56	if (x == s) {                   // wait was cancelled

57	clean(t, s);

58	return null;

59

}

60

61	if (!s.isOffList()) {           // not already unlinked

62	advanceHead(t, s);          // unlink if head

63	if (x != null)              // and forget fields

64	s.item = s;

65	s.waiter = null;

66

}

67	return (x != null)? x : e;

68

69	} else {                            // complementary-mode

70	QNode m = h.next;               // node to fulfill

71	if (t != tail || m == null || h != head)

72	continue;                   // inconsistent read

73

74	Object x = m.item;

75	if (isData == (x != null) ||    // m already fulfilled

76	x == m ||                   // m cancelled

77	!m.casItem(x, e)) {         // lost CAS

78	advanceHead(h, m);          // dequeue and retry

79

continue;

80

}

81

82	advanceHead(h, m);              // successfully fulfilled

83	LockSupport.unpark(m.waiter);

84	return (x != null)? x : e;

85

}

86

}

87

}