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

ConcurrentLinkedQueue的实现和非阻塞同步算法前言 本文写给对ConcurrentLinkedQueue的实现和非阻塞同步算法的实现原理有一定了解，但缺少实践经验的朋友，文中包括了实战中的尝试、所走的弯路，经验和教训。 背景介绍 独自负责一个聊天系统的服务端，因为一些原因，我没使用现成的开源框架，网络那块直接使用AIO，收数据时，因为只会从 channel里过来，所以不需要考虑同步问题；但是发送数据时，因为有聊天消息的转发，所以必需处理这个同步问题。AIO中，是处理完一个注册的操作 后，再执行我们定义的方法，此时，如果还有数据需要写，则继续注册写操作，如果没有则不注册；提交数据时，如果当前没有注册写操作，则注册一个，否则仅提 交（此时再注册则会报异常）。这样，需要同步的点就是：如何知道当前还有没有数据需要发送（因为其它线程也可以提交数据到此处），和如何知道此次提交时， 有没有注册写操作。总之，要完成：有数据要发送时，必需有写操作被注册，并且只能注册一次；没有数据时，不能有写操作被注册。 问题分析 经过分析，上面的问题，可以抽象成：我需要知道当往队列里插入一条数据之前，该队列是否为空，如果为空则应该注册新的写操作。当从队列里取出一条数据后，该队列是否为非空，如果非空则应该继续注册写操作。（本文之后以“关注的操作”来表示这种场景下的插入或取出操作） 目前的问题是，我使用的队列是ConcurrentLinkedQueue，但是它的取出数据的方法，没有返回值告诉我们从队列里取出数据之后队列 是否为空，如果是使用size或peek方法来执行判断，那就必需加锁了，否则在拿到队列大小时，可能队列大小已经变化了。所以我首先想到的是，如何对该 队列进行改造，让它提供该信息。 注意：这里指的不是当次而是之后，所以如果我们使用队列的peek()方法返回null，就知道队列是否为空，但是不知道之后是否为空 ，并且，当关注的操作发生时，在插入或取出操作的返回值里告知此信息，来指导是否继续注册写操作。 用锁实现 如果使用锁的话很容易处理，用锁同步插入和取出方法，下面是锁实现的参考： 1 public E poll() { 2 synchronized (this) { 3 E re = q.poll(); 4 // 获取元素后，队列是空，表示是我关注的操作 5 if (q.peek() == null) { 6 7 } 8 return re; 9 } 10 } 11 12 public void offer(E e) { 13 synchronized (this) { 14 // 插入元素前，队列是空，表示是我关注的操作 15 if (q.peek() == null) { 16 17 } 18 q.offer(e); 19 } 20 } 但因为是服务端，我想用非阻塞同步算法来实现。 尝试方案一 我第一次想到的改造办法是，将head占位节点改成固定的，头节点移动时，只将head节点的next指向新的节点，在插入数据时，如果是在 head节点上成功执行的该操作，那么该插入就是关注的的操作；在取出时，如果将head节点的next置为了null，那么该取出就是关注的操作（?因 为之前的占位节点是变化的，所以没法比较，必需用同步，现在是固定的了，所以可以直接与head节点进行比较?）。如此一来，问题好像被解决了。改造完之 后，出于严谨，我仔细读了一遍代码，发现引入了新的问题，我的取出操作是这样写的 1 /** 2 * @author trytocatch@163.com 3 */ 4 public E poll(){ 5 for(;;){ 6 Node n = head.nextRef.get();//head指向固定的head节点，为final 7 if(n == null) 8 return null; 9 Node m = n.nextRef.get(); 10 if(pareAndSet(n,m){ 11 if(m==null)12 /*此时为关注的操作（为了简化代码显示，不将该信息当作返回值返回了，仅注释）13 return n.itemRef.get(); 14 } 15 } 16 } 这里有一个