ConcurrentLinkedQueue的实现和非阻塞同步算法

前言本文写给对 ConcurrentLinkedQueue 的实现和非阻塞同步算法的实现原理有一定了解 但缺少实践经验的朋友，文中包括了实战中的尝试、所走的弯路，经验和教训。背景介绍本人独自负责一个聊天系统的服务端，因为一些原因，我没使用现成的开源框架，网络 那块直接使用AIO,收数据时，因为只会从channel里过来，所以不需要考虑同步问题；但 是发送数据时，因为有聊天消息的转发，所以必需处理这个同步问题oAIO中，是处理完一 个注册的操作 后，再执行我们定义的方法，此时，如果还有数据需要写，则继续注册写操 作，如果没有则不注册；提交数据时，如果当前没有注册写操作，则注册一个，否则仅提 交 (此时再注册则会报异常)。这样，需要同步的点就是：如何知道当前还有没有数据需要发 送(因为其它线程也可以提交数据到此处)，和如何知道此次提交时， 有没有注册写操作。 总之，要完成：有数据要发送时，必需有写操作被注册，并且只能注册一次；没有数据时， 不能有写操作被注册。问题分析经过分析，上面的问题，可以抽象成：我需要知道当往队列里插入一条数据之前，该队 列是否为空，如果为空则应该注册新的写操作。当从队列里取出一条数据后，该队列是否为 非空，如果非空则应该继续注册写操作。(本文之后以“关注的操作”来表示这种场景下的插 入或取出操作)目前的问题是，我使用的队列是ConcurrentLinkedQueue,但是它的取出数据的方法， 没有返回值告诉我们从队列里取出数据之后队列是否为空，如果是使用size或peek方法来 执行判断，那就必需加锁了，否则在拿到队列大小时，可能队列大小已经变化了。所以我首 先想到的是，如何对该 队列进行改造，让它提供该信息。注意：这里指的不是当次而是之后，所以如果我们使用队列的peek()方法返回null,就 知道队列是否为空，但是不知道之后是否为空 ，并且，当关注的操作发生时，在插入或取 出操作的返回值里告知此信息，来指导是否继续注册写操作。用锁实现 如果使用锁的话很容易处理，用锁同步插入和取出方法，下面是锁实现的参考：1 public E poll() 2 synchronized (this) 3 Ere = q.poll();4 /获取元素后，队列是空，表示是我关注的操作5 if(q.peek() = null) 678return re;910 1112 public void offer(E e) 13synchronized (this) 14/ 插入元素前，队列是空，表示是我关注的操作15if (q.peek() = null) 161718q.offer(e);1920 但因为是服务端，我想用非阻塞同步算法来实现。八乞、r 、尝试方案一我第一次想到的改造办法是，将 head 占位节点改成固定的，头节点移动时，只将 head 节点的 next 指向新的节点，在插入数据时，如果是在 head 节点上成功执行的该操作，那么 该插入就是关注的的操作；在取出时，如果将head节点的next置为了 null,那么该取出就 是关注的操作（因 为之前的占位节点是变化的，所以没法比较，必需用同步，现在是固定 的了，所以可以直接与head节点进行比较）。如此一来，问题好像被解决了。改造完之后, 出于严谨，我仔细读了一遍代码，发现引入了新的问题，我的取出操作是这样写的1 /*2 * author trytocatch163.com3 */4 public E poll（）5for(;)6Node n = head.nextRef.get();/head 指向固定的 head 节点，为 final7if(n = null)8return null;9Node m = n.nextRef.get();10if(head.next.compareAndSet(n,m)11if(m=null)；12/*此时为关注的操作（为了简化代码显示，不将该信息当作返回值返回了，仅注释）*/13return n.itemRef.get();141516 这里有一个致命的问题：如果m为null，在CAS期间，插入了新节点，n的next由null 变成了非null，紧接着又把head的next更新为了 null,那么链就断了，该方法还存在一些 其它的问题，如当队列为空的时候，尾节点指向了错误的位置，本应该是head的。我认为 最根本的原因在于，head不能设为固定的，否则会引发一堆问题。第一次尝试宣告失败。八匕、卜.尝试方案二这次我尝试将head跟tail两个引用包装成一个对象,然后对这个对象进行CAS更新(这 仅为一处理论上的尝试，因为从性能上面来讲，已经大打折 扣了，创建了很多额外的对象)， 如果 head 跟 tail 指向了同一个节点,则认为队列是空的,根据此信息来判断一个操作是不 是关注的操作。但该尝试仅停留 在了理论分析阶段,期间发现了一些小问题,没法解决, 后来我发现,我把 ConcurrentLinkedQueue 原本可以分成两步执行的插入和取出操作 (更新 节点的 next 或 item 引用,然后再更新 head 或 tail 引用),变成了必需一步完成, ConcurrentLinkedQueue 尚不能一步 完成,我何德何能,可将它们一步完成？所以直接放弃 了。解决方案一经过两次的失败尝试,我几乎绝望了,我怀疑这是不是不能判断出是否为关注的操作。 因为是在做项目,周末已经过去了,不能再搞这些“研究”了,所以我退而求其次,想了 个不太漂亮的办法,在队列外部维护一个变量,记录队列有多大,在 插入或取出后,更新 该变量，使用的是AtomicInteger,如果更新时，将变量从1变成0,或从0变成了 1,就认 为该插入或取出为关注的操作。I private AtomicInteger size = new AtomicInteger(0);2 public E poll() 3 E re = q.poll();4 if (re = null)5 return null;6 for(int old;)7 old = size.get();8 if(size.compareAndSet(old,old-1)9 / 获取元素后,队列是空,表示是我关注的操作10 if(old =1)II12 13 break;14 15 16 return re;17 1819 public void offer(E e) 20 q.offer(e);21 for(int old;)22 old = size.get();23 if(size.compareAndSet(old,old+1)24 / 插入元素前,队列是空,表示是我关注的操作25 if(old = 0)262728break;29 30 31 此时，也许细心的朋友会问，因为没有使用锁，这个变量并不能真实反映队列的大 小， 也就不能确定它是不是关注的操作。没错，是不能真实反映，但是，我获取关注的操作的目 的是用来指导我：该不该注册新的写操作，该变量的值变化就能提供 正确的指导，所以， 同样是正确的，只不过途径不同而已。理论上的分析和后来的项目正确运行都印证了该方法 的正确性。解决方案二因为上面的方法额外加了一次lock-free级别的CAS操作，我心里总不太舒服，空余时 间总在琢磨，真的就没有办法，在不增加额外lock-free级别CAS开支的情况下，知晓一个 操作是不是关注的操作？后来经分析，如果要知晓是不是关注的操作，跟两个数据有关，真实的头节点跟尾节点 (不同于head跟tail，因为它们是滞后的，之前将它们包装成一个对象就是犯了该错误)， ConcurrentLinkedQueue的实现中，这两个节点是没有竞争的，一个是更新item,一个是更新 next，必 需得让他们竞争同一个东西，才能解决该问题，于是我想到了一个办法，取出完 成后，如果该节点的next为null，就将其用CAS置为一个特殊的值，若成功则认为是关注 的操作；插入成功后，如果 next 被替换掉的值不是 null 而是这个特殊值，那么该插入也为 关注的操作。这仅增加了一次wait-free级别的CAS操作(取出后的那次CAS)，perfect!因为 ConcurrentLinkedQueue 的很多变量、内部类都是私有的，没法通过继承来改造， 没办法，只得自行实现。对于队列里使用的Node，实现的方式有很多种，可以使用AtomicReference、AtomicReferenceFieldUpdater 来实现，如果你愿意的话，甚至是像ConcurrentLinkedQueue 一样，用 sun.misc.Unsafe 来实现(注意：一般来说， sun 包下的类是 不推荐使用 的)，各有优缺点吧，所以我就不提供该队列的具体实现了，下面给出在 ConcurrentLinkedQueue (版本：1.7.0_10)基础上进行的改造，供参考。注意，如果需要用 到 size 等方法，因为特殊值的引入，影响了之前的判断逻辑，应重新编写。1 /*2 * author trytocatch163.com3 */4 private static final Node MARK_NODE = new Node(null);55 public boolean offer(E e) 6 checkNotNull(e);7 final Node newNode = new Node(e);98 for (Node t = tail, p = t;) 9 Node q = p.next;10 if (q = null | q = MARK_NODE) /修改 1：加入条件：或 q = MARK_NODE11 / p is last node12 if (p.casNext(q, newNode) /修改 2：将 null 改为 q1516171819202122232425262728293031323334353637383940414243444546474849505152535455565758/ Successful CAS is the linearization point / for e to become an element of this queue, / and for newNode to become "live".if (q = MARK_NODE)修改 3:;/此时为关注的操作(为了简化代码显示，仅注释) if (p != t) / hop two nodes at a timecasTail(t, newNode); / Failure is OK. return true;/ Lost CAS race to another thread; r