blocks|key|137648|text|这就是了：|type|unstyled|depth|inlineStyleRanges|entityRanges|data|137649|public+class+FunkyQueue<T>+{
++++private+final+AtomicReference<Node<T>>+_tail+=+new+AtomicReference<Node<T>>();

++++public+void+offer(T+t)+{
++++++++while(true)+{
++++++++++++Node<T>+tail+=+_tail.get();
++++++++++++Node<T>+newTail+=+new+Node<T>(t,+tail);
++++++++++++if(_tail.compareAndSet(tail,+newTail))+{
++++++++++++++++break;
++++++++++++}
++++++++}
++++}

++++public+List<T>+takeAll()+{
++++++++Node<T>+tail+=+_tail.getAndSet(null);

++++++++LinkedList<T>+list+=+new+LinkedList<T>();
++++++++while(tail+!=+null)+{
++++++++++++list.addFirst(tail.get());
++++++++++++tail+=+tail.getPrevious();
++++++++}

++++++++return+list;
++++}

++++private+static+final+class+Node<T>
++++{
++++++++private+final+T+_obj;
++++++++private+Node<T>+_prev;

++++++++private+Node(T+obj,+Node<T>+prev)+{
++++++++++++_obj+=+obj;
++++++++++++_prev+=+prev;++++++++++++
++++++++}

++++++++public+T+get()+{
++++++++++++return+_obj;
++++++++}

++++++++public+Node<T>+getPrevious()+{
++++++++++++return+_prev;
++++++++}
++++}
}|code-block|syntax|javascript|137650|entityMap^0|0|0^^$0|@$1|2|3|4|5|6|7|I|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|J|8|@]|9|@]|A|$E|F]]|$1|G|3|-4|5|6|7|K|8|@]|9|@]|A|$]]]|H|$]]

Here you go:

<pre><code>public class FunkyQueue&lt;T&gt; {
 private final AtomicReference&lt;Node&lt;T&gt;&gt; _tail = new AtomicReference&lt;Node&lt;T&gt;&gt;();

 public void offer(T t) {
 while(true) {
 Node&lt;T&gt; tail = _tail.get();
 Node&lt;T&gt; newTail = new Node&lt;T&gt;(t, tail);
 if(_tail.compareAndSet(tail, newTail)) {
 break;
 }
 }
 }

 public List&lt;T&gt; takeAll() {
 Node&lt;T&gt; tail = _tail.getAndSet(null);

 LinkedList&lt;T&gt; list = new LinkedList&lt;T&gt;();
 while(tail != null) {
 list.addFirst(tail.get());
 tail = tail.getPrevious();
 }

 return list;
 }

 private static final class Node&lt;T&gt;
 {
 private final T _obj;
 private Node&lt;T&gt; _prev;

 private Node(T obj, Node&lt;T&gt; prev) {
 _obj = obj;
 _prev = prev; 
 }

 public T get() {
 return _obj;
 }

 public Node&lt;T&gt; getPrevious() {
 return _prev;
 }
 }
}
</code></pre>

blocks|key|137741|text|在给定：nice+implementation的情况下，offer()和takeAll()都需要一个CAS。|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|137742|Problem：长时间takeAll()执行，因为它需要在相反方向上完全遍历单链表。|BOLD|137743|对于前面提到的数字(N+~+100K)，两个级别就足够了，从而将takeAll()中的步骤数减少到大约150步。|137744|基于前面提到的实现，Node类：|137745|public+static+final+class+Node<T>+{

++++private+final+T+value;
++++private+Node<T>+prev,+prevL1,+prevL2;
++++private+Node<T>+next,+nextL1,+nextL2;

++++private+Node(T+obj,+Node<T>+prev,+long+c)+{
++++++++value+=+obj;
++++++++this.prev+=+prev;++
++++++++//+level+1+to+skip+64+nodes,+level+2+to+skip+64%5E2+nodes
++++++++//+c+is+a+value+from+some+global+addition+counter,+that
++++++++//+is+not+required+to+be+atomic+with+`offer()`
++++++++prevL1+=+(c+&+(64+-+1)+==+0)+?+prev+:+prev.prevL1;
++++++++prevL2+=+(c+&+(64+*+64+-+1)+==+0)+?+prev+:+prev.prevL2;
++++}

++++public+T+get()+{
++++++++return+value;
++++}

++++public+Node<T>+findHead()+{
++++++++//+see+below
++++}

++++public+Node<T>+next()+{
++++++++//+see+below
++++}
}|code-block|syntax|javascript|137746|FunkyQueue#offer()方法：|137747|public+void+offer(T+t)+{
++++long+c+=+counter.incrementAndGet();++
++++for(;;)+{
++++++++Node<T>+oldTail+=+tail.get();
++++++++Node<T>+newTail+=+new+Node<T>(t,+oldTail,+c);
++++++++if+(tail.compareAndSet(oldTail,+newTail))+
++++++++++++break;
++++}
}|137748|FunkyQueue#takeAll()现在将返回列表的头部：|137749|public+Node<T>+takeAll()+{
++++return+tail.getAndSet(null).findHead();
}|137750|137751|private+Node<T>+findHead()+{

+++++Node<T>+n+=+this;
+++++while+(n.prevL2+!=+null)+{++//+<-+traverse+back+on+L2,+assigning+`next`+nodes
+++++++++n.prevL2.nextL2+=+n;
+++++++++n+=+n.prevL2;+
+++++}
+++++while+(n.prevL1+!=+null)+{++//+<-+the+same+for+L1
+++++++++n.prevL1.nextL1+=+n;
+++++++++n+=+n.prev1;
+++++}
+++++while+(n.prev+!=+null)+{++++//+<-+the+same+for+L0
+++++++++n.prev.next+=+n;
+++++++++n+=+n.prev;
+++++}
+++++return+n;
}|137752|最后是Node#next()|137753|public+Node<T>+next()+{

++++if+(this.next+==+null+&&+this.nextL1+==+null+&&+this.nextL2+==+null)+++++++
++++++++throw+new+IllegalStateException("No+such+element");

++++Node<T>+n;
++++if+(this.next+==+null)+{+++++++++//+L0+is+not+traversed+yet
++++++++if+(this.nextL1+==+null)+{+++//+the+same+for+L1
++++++++++++n+=+this.nextL2;+++++++++//+step+forward+on+L2
++++++++++++while+(n+!=+this)+{++++++//+traverse+on+L1
++++++++++++++++n.prevL1.nextL1+=+n;
++++++++++++++++n+=+n.prevL1;
++++++++++++}
++++++++}++
++++++++n+=+this.nextL1;+++++++++++++//+step+forward+on+L1
++++++++while+(n+!=+this)+{++++++++++//+traverse+on+L0
++++++++++++n.prev.next+=+n;
++++++++++++n+=+n.prev;
++++++++}
++++}
++++return+this.next;
}|137754|我想大意已经很清楚了。应用一些重构，可以使Node#findHead()和FunkyQueue#takeAll()在O(log+)中运行，而Node#next()在O(1)中运行。|137755|137756|137757|entityMap|0|LINK|mutability|MUTABLE|url|https://stackoverflow.com/a/29779790/3459206^0|S|7|10|9|4|J|0|0|0|7|B|9|0|W|9|0|A|4|0|0|0|I|0|0|0|K|0|0|0|0|3|B|0|0|L|F|11|K|1Y|B|0|0|0^^$0|@$1|2|3|4|5|6|7|1I|8|@$9|1J|A|1K|B|C]|$9|1L|A|1M|B|C]]|D|@$9|1N|A|1O|1|1P]]|E|$]]|$1|F|3|G|5|6|7|1Q|8|@$9|1R|A|1S|B|H]|$9|1T|A|1U|B|C]]|D|@]|E|$]]|$1|I|3|J|5|6|7|1V|8|@$9|1W|A|1X|B|C]]|D|@]|E|$]]|$1|K|3|L|5|6|7|1Y|8|@$9|1Z|A|20|B|C]]|D|@]|E|$]]|$1|M|3|N|5|O|7|21|8|@]|D|@]|E|$P|Q]]|$1|R|3|S|5|6|7|22|8|@$9|23|A|24|B|C]]|D|@]|E|$]]|$1|T|3|U|5|O|7|25|8|@]|D|@]|E|$P|Q]]|$1|V|3|W|5|6|7|26|8|@$9|27|A|28|B|C]]|D|@]|E|$]]|$1|X|3|Y|5|O|7|29|8|@]|D|@]|E|$P|Q]]|$1|Z|3|-4|5|6|7|2A|8|@]|D|@]|E|$]]|$1|10|3|11|5|O|7|2B|8|@]|D|@]|E|$P|Q]]|$1|12|3|13|5|6|7|2C|8|@$9|2D|A|2E|B|C]]|D|@]|E|$]]|$1|14|3|15|5|O|7|2F|8|@]|D|@]|E|$P|Q]]|$1|16|3|17|5|6|7|2G|8|@$9|2H|A|2I|B|C]|$9|2J|A|2K|B|C]|$9|2L|A|2M|B|C]]|D|@]|E|$]]|$1|18|3|-4|5|6|7|2N|8|@]|D|@]|E|$]]|$1|19|3|-4|5|6|7|2O|8|@]|D|@]|E|$]]|$1|1A|3|-4|5|6|7|2P|8|@]|D|@]|E|$]]]|1B|$1C|$5|1D|1E|1F|E|$1G|1H]]]]

Given: <a href="https://stackoverflow.com/a/29779790/3459206">nice implementation</a>, that requires a single CAS for both <code>offer()</code> and <code>takeAll()</code>. 

Problem: long <code>takeAll()</code> execution, since it needs full traversal of single-linked list in opposite direction.

Solution: create additional skip levels on nodes. For mentioned figures (N ~ 100K) two levels would be enough, thus reducing number of steps in <code>takeAll()</code> to ~150.

Based on mentioned implementation, <code>Node</code> class:

<pre><code>public static final class Node&lt;T&gt; {

 private final T value;
 private Node&lt;T&gt; prev, prevL1, prevL2;
 private Node&lt;T&gt; next, nextL1, nextL2;

 private Node(T obj, Node&lt;T&gt; prev, long c) {
 value = obj;
 this.prev = prev; 
 // level 1 to skip 64 nodes, level 2 to skip 64^2 nodes
 // c is a value from some global addition counter, that
 // is not required to be atomic with `offer()`
 prevL1 = (c &amp; (64 - 1) == 0) ? prev : prev.prevL1;
 prevL2 = (c &amp; (64 * 64 - 1) == 0) ? prev : prev.prevL2;
 }

 public T get() {
 return value;
 }

 public Node&lt;T&gt; findHead() {
 // see below
 }

 public Node&lt;T&gt; next() {
 // see below
 }
}
</code></pre>

<code>FunkyQueue#offer()</code> method:

<pre><code>public void offer(T t) {
 long c = counter.incrementAndGet(); 
 for(;;) {
 Node&lt;T&gt; oldTail = tail.get();
 Node&lt;T&gt; newTail = new Node&lt;T&gt;(t, oldTail, c);
 if (tail.compareAndSet(oldTail, newTail)) 
 break;
 }
}
</code></pre>

<code>FunkyQueue#takeAll()</code> would now return the head of the list:

<pre><code>public Node&lt;T&gt; takeAll() {
 return tail.getAndSet(null).findHead();
}
</code></pre>

It calls <code>Node#findHead()</code>, which now can use skip levels to speed up backward traversal:

<pre><code>private Node&lt;T&gt; findHead() {

 Node&lt;T&gt; n = this;
 while (n.prevL2 != null) { // &lt;- traverse back on L2, assigning `next` nodes
 n.prevL2.nextL2 = n;
 n = n.prevL2; 
 }
 while (n.prevL1 != null) { // &lt;- the same for L1
 n.prevL1.nextL1 = n;
 n = n.prev1;
 }
 while (n.prev != null) { // &lt;- the same for L0
 n.prev.next = n;
 n = n.prev;
 }
 return n;
}
</code></pre>

And finally, <code>Node#next()</code>:

<pre><code>public Node&lt;T&gt; next() {

 if (this.next == null &amp;&amp; this.nextL1 == null &amp;&amp; this.nextL2 == null) 
 throw new IllegalStateException("No such element");

 Node&lt;T&gt; n;
 if (this.next == null) { // L0 is not traversed yet
 if (this.nextL1 == null) { // the same for L1
 n = this.nextL2; // step forward on L2
 while (n != this) { // traverse on L1
 n.prevL1.nextL1 = n;
 n = n.prevL1;
 }
 } 
 n = this.nextL1; // step forward on L1
 while (n != this) { // traverse on L0
 n.prev.next = n;
 n = n.prev;
 }
 }
 return this.next;
}
</code></pre>

I suppose the main idea is clear. Applying some refactoring, it is possible to make <code>Node#findHead()</code> and thus <code>FunkyQueue#takeAll()</code> operate in O(log N), and <code>Node#next()</code> in O(1) at average.

<hr>

P.S. if one notices some error or bad grammar, please edit.

blocks|key|2917914|text|ConcurrentLinkedQueue使用Michael+&+Scott+algorithm，并且可以调整以提供此方法。返回的集合将是用于遍历的已移除节点的不可修改视图。它看起来就像，|type|unstyled|depth|inlineStyleRanges|entityRanges|offset|length|data|2917915|public+Collection<E>+drain()+{
++for+(;;)+{
++++Node<E>+h+=+head;
++++Node<E>+t+=+tail;

++++if+(h+==+t)+{
++++++return+Collections.emptyList();
++++}+else+if+(casHead(h,+t))+{
++++++return+new+CollectionView<E>(h,+t);
++++}
++}
}|code-block|syntax|javascript|2917916|派生一个集合并不是很有趣，所以我使用implemented+Mozes+&+Shavit+algorithm获取一个乐观队列(一种更快的替代方法)。这是通过回退舞台来组合并发添加来减少由多个生产者引起的争用而得到增强的。|2917917|entityMap|0|LINK|mutability|MUTABLE|url|https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ConcurrentLinkedQueue.html|1|http://www.cs.rochester.edu/~scott/papers/1996_PODC_queues.pdf|2|https://github.com/ben-manes/caffeine/blob/520ee5ba17778d503ba1fa753ddc43d7bb59ca40/caffeine/src/main/java/com/github/benmanes/caffeine/ConcurrentLinkedLazyQueue.java|3|http://people.csail.mit.edu/edya/publications/OptimisticFIFOQueue-journal.pdf^0|0|L|0|N|P|1|0|0|I|B|2|U|O|3|0^^$0|@$1|2|3|4|5|6|7|Y|8|@]|9|@$A|Z|B|10|1|11]|$A|12|B|13|1|14]]|C|$]]|$1|D|3|E|5|F|7|15|8|@]|9|@]|C|$G|H]]|$1|I|3|J|5|6|7|16|8|@]|9|@$A|17|B|18|1|19]|$A|1A|B|1B|1|1C]]|C|$]]|$1|K|3|-4|5|6|7|1D|8|@]|9|@]|C|$]]]|L|$M|$5|N|O|P|C|$Q|R]]|S|$5|N|O|P|C|$Q|T]]|U|$5|N|O|P|C|$Q|V]]|W|$5|N|O|P|C|$Q|X]]]]

<a href="https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ConcurrentLinkedQueue.html" rel="nofollow">ConcurrentLinkedQueue</a> uses the <a href="http://www.cs.rochester.edu/~scott/papers/1996_PODC_queues.pdf" rel="nofollow">Michael &amp; Scott algorithm</a> and could be adapted to provide this method. The returned collection would be an unmodifiable view of the removed nodes for traversal. That would look something like,

<pre><code>public Collection&lt;E&gt; drain() {
 for (;;) {
 Node&lt;E&gt; h = head;
 Node&lt;E&gt; t = tail;

 if (h == t) {
 return Collections.emptyList();
 } else if (casHead(h, t)) {
 return new CollectionView&lt;E&gt;(h, t);
 }
 }
}
</code></pre>

Forking a collection isn't very fun, so instead I <a href="https://github.com/ben-manes/caffeine/blob/520ee5ba17778d503ba1fa753ddc43d7bb59ca40/caffeine/src/main/java/com/github/benmanes/caffeine/ConcurrentLinkedLazyQueue.java" rel="nofollow">implemented</a> the <a href="http://people.csail.mit.edu/edya/publications/OptimisticFIFOQueue-journal.pdf" rel="nofollow">Mozes &amp; Shavit algorithm</a> for an optimistic queue (a faster alternative). This is enhanced with a backoff arena to combine concurrent additions to reduce contention caused by multiple producers.

Non-Blocking Concurrent Queue with offer and flush

I'd need an unbounded non-blocking concurrent queue with basically only 2 operations:

<ul>
<li>offer : atomically inserts the specified item at the tail of this queue;</li>
<li>flush : takes all the items present in that moment in the queue and start processing them one by one following the insertion order. More specifically what has to be atomic is only this "takeAll" operation that will be the very first operation of the flush. All items offered to the queue after the takeAll will be inserted and then processed only by a another subsequent flush.</li>
</ul>

The goal is the consumer has a single CAS operation on the takeAll and can then iterate the elements in the list without going through a CAS operation per read. Further we already own the Node (Entry) as this is needed to store some other immutable state. The new node can take the HEAD as a constructor argument, creating a single direction linked list.

Does exist in literature a queue with these characteristics?

Non-Blocking Concurrent Queue with offer and flush

具有提供和刷新功能的非阻塞并发队列我需要一个无界的非阻塞并发队列，基本上只有2个操作：offer :原子地将指定项插入到此队列的尾部；flush:获取队列中该时刻出现的所有项，并按照插入顺序逐个开始处理它们。更具体地说，必须是原子的只是这个"takeAll“操作，它将是刷新的第一个操作。在takeAll之后提供给队列的...

问具有提供和刷新功能的非阻塞并发队列
EN

社区

活动

资源

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问具有提供和刷新功能的非阻塞并发队列EN

社区

活动

资源

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问具有提供和刷新功能的非阻塞并发队列
EN