java中快如闪电的线程间通讯-mile米乐体育

这个故事源自一个很简单的想法：创建一个对开发人员友好的、简单轻量的线程间通讯框架，完全不用锁、同步器、信号量、等待和通知，在java里开发一个轻量、无锁的线程内通讯框架；并且也没有队列、消息、事件或任何其他并发专用的术语或工具。

只用普通的老式java接口实现pojo的通讯。

它可能跟akka的类型化actor类似，但作为一个必须超级轻量，并且要针对单台多核计算机进行优化的新框架，那个可能有点过了。

当actor跨越不同jvm实例（在同一台机器上，或分布在网络上的不同机器上）的进程边界时，akka框架很善于处理进程间的通讯。

但对于那种只需要线程间通讯的小型项目而言，用akka类型化actor可能有点儿像用牛刀杀鸡，不过类型化actor仍然是一种理想的实现方式。

我花了几天时间，用动态代理，阻塞队列和缓存线程池创建了一个mile米乐体育的解决方案。

图一是这个框架的高层次架构：

图一: 框架的高层次架构

spsc队列是指单一生产者/单一消费者队列。mpsc队列是指多生产者/单一消费者队列。

派发线程负责接收actor线程发送的消息，并把它们派发到对应的spsc队列中去。

接收到消息的actor线程用其中的数据调用相应的actor实例中的方法。借助其他actor的代理，actor实例可以将消息发送到mpsc队列中，然后消息会被发送给目标actor线程。

我创建了一个简单的例子来测试，就是下面这个打乒乓球的程序：

public interface playera (   void pong(long ball); //发完就忘的方法调用  } public interface playerb {      void ping(playera playera, long ball); //发完就忘的方法调用  }     public class playeraimpl implements playera {       @override       public void pong(long ball) {       }     } public class playerbimpl implements playerb {      @override       public void ping(playera playera, long ball) {         playera.pong(ball);       }     } public class pingpongexample {      public void testpingpong() {     // 管理器隐藏了线程间通讯的复杂性     // 控制actor代理，actor实现和线程       actormanager manager = new actormanager();     // 在管理器内注册actor实现      manager.registerimpl(playeraimpl.class);         manager.registerimpl(playerbimpl.class);     //创建actor代理。代理会将方法调用转换成内部消息。      //会在线程间发给特定的actor实例。         playera playera = manager.createactor(playera.class);         playerb playerb = manager.createactor(playerb.class);         for(int i = 0; i < 1000000; i  ) {            playerb.ping(playera, i);         }     }

经过测试，速度大约在每秒500,000 次乒/乓左右；还不错吧。然而跟单线程的运行速度比起来，我突然就感觉没那么好了。在 单线程中运行的代码每秒速度能达到20亿 (2,681,850,373)！

居然差了5,000 多倍。太让我失望了。在大多数情况下，单线程代码的效果都比多线程代码更高效。

我开始找原因，想看看我的乒乓球运动员们为什么这么慢。经过一番调研和测试，我发现是阻塞队列的问题，我用来在actor间传递消息的队列影响了性能。

图 2: 只有一个生产者和一个消费者的spsc队列

所以我发起了一场竞赛，要将它换成java里最快的队列。我发现了nitsan wakart的 博客 。他发了几篇文章介绍单一生产者/单一消费者(spsc)无锁队列的实现。这些文章受到了martin thompson的演讲 终极性能的无锁算法的启发。

跟基于私有锁的队列相比，无锁队列的性能更优。在基于锁的队列中，当一个线程得到锁时，其它线程就要等着锁被释放。而在无锁的算法中，某个生产者线程生产消息时不会阻塞其它生产者线程，消费者也不会被其它读取队列的消费者阻塞。

在martin thompson的演讲以及在nitsan的博客中介绍的spsc队列的性能简直令人难以置信—— 超过了100m ops/sec。比jdk的并发队列实现还要快10倍 (在4核的 intel core i7 上的性能大约在 8m ops/sec 左右)。

我怀着极大的期望，将所有actor上连接的链式阻塞队列都换成了无锁的spsc队列。可惜，在吞吐量上的性能测试并没有像我预期的那样出现大幅提升。不过很快我就意识到，瓶颈并不在spsc队列上，而是在多个生产者/单一消费者(mpsc)那里。

用spsc队列做mpsc队列的任务并不那么简单；在做put操作时，多个生产者可能会覆盖掉彼此的值。spsc 队列就没有控制多个生产者put操作的代码。所以即便换成最快的spsc队列，也解决不了我的问题。

为了处理多个生产者/单一消费者的情况，我决定启用lmax disruptor ——一个基于环形缓冲区的高性能进程间消息库。

图3: 单一生产者和单一消费者的lmax disruptor

借助disruptor，很容易实现低延迟、高吞吐量的线程间消息通讯。它还为生产者和消费者的不同组合提供了不同的用例。几个线程可以互不阻塞地读取环形缓冲中的消息：

图 4: 单一生产者和两个消费者的lmax disruptor    

下面是有多个生产者写入环形缓冲区，多个消费者从中读取消息的场景。

图 5: 两个生产者和两个消费者的lmax disruptor

经过对性能测试的快速搜索，我找到了 三个发布者和一个消费者的吞吐量测试。 这个真是正合我意，它给出了下面这个结果：

在3 个生产者/1个 消费者场景下， disruptor要比linkedblockingqueue快两倍多。然而这跟我所期望的性能上提升10倍仍有很大差距。

这让我觉得很沮丧，并且我的大脑一直在搜寻mile米乐体育的解决方案。就像命中注定一样，我最近不在跟人拼车上下班，而是改乘地铁了。突然灵光一闪，我的大脑开始将车站跟生产者消费者对应起来。在一个车站里，既有生产者（车和下车的人），也有消费者（同一辆车和上车的人）。

我创建了 railway类，并用atomiclong追踪从一站到下一站的列车。我先从简单的场景开始，只有一辆车的铁轨。

public class railway {    private final train train = new train();    // stationno追踪列车并定义哪个车站接收到了列车  private final atomicinteger stationindex = new atomicinteger(); // 会有多个线程访问这个方法，并等待特定车站上的列车  public train waittrainonstation(final int stationno) {     while (stationindex.get() % stationcount != stationno) {     thread.yield(); // 为保证高吞吐量的消息传递，这个是必须的。                    //但在等待列车时它会消耗cpu周期     }      // 只有站号等于stationindex.get() % stationcount时，这个忙循环才会返回     return train;  } // 这个方法通过增加列车的站点索引将这辆列车移到下一站   public void sendtrain() {     stationindex.getandincrement();    }   }

为了测试，我用的条件跟在disruptor性能测试中用的一样，并且也是测的spsc队列——测试在线程间传递long值。我创建了下面这个train类，其中包含了一个long数组：

public class train {      //      public static int capacity = 2*1024;   private final long[] goodsarray; // 传输运输货物的数组    private int index;    public train() {          goodsarray = new long[capacity];       }   public int goodscount() { //返回货物数量       return index;      }      public void addgoods(long i) { // 向列车中添加条目       goodsarray[index  ] = i;      }      public long getgoods(int i) { //从列车中移走条目       index--;       return goodsarray[i];      }     }

然后我写了一个简单的测试 ：两个线程通过列车互相传递long值。

图 6: 使用单辆列车的单一生产者和单一消费者railway

public void testrailway() {      final railway railway = new railway();       final long n = 20000000000l;       //启动一个消费者进程    new thread() {        long lastvalue = 0;    @override       public void run() {         while (lastvalue < n) {           train train = railway.waittrainonstation(1); //在#1站等列车       int count = train.goodscount();           for (int i = 0; i < count; i  ) {             lastvalue = train.getgoods(i); // 卸货          }           railway.sendtrain(); //将当前列车送到第一站       }        }      }.start();  final long start = system.nanotime(); long i = 0;    while (i < n) {      train train = railway.waittrainonstation(0); // 在#0站等列车      int capacity = train.getcapacity();      for (int j = 0; j < capacity; j  ) {        train.addgoods((int)i  ); // 将货物装到列车上   }      railway.sendtrain();  if (i % 100000000 == 0) { //每隔100m个条目测量一次性能      final long duration = system.nanotime() - start;         final long ops = (i * 1000l * 1000l * 1000l) / duration;         system.out.format("ops/sec = %,d\n", ops);         system.out.format("trains/sec = %,d\n", ops / train.capacity);         system.out.format("latency nanos = %.3f%n\n",      duration / (float)(i) * (float)train.capacity);       }      }     }

在不同的列车容量下运行这个测试，结果惊着我了：

在列车容量达到32,768时，两个线程传送消息的吞吐量达到了767,028,751 ops/sec。比nitsan博客中的spsc队列快了几倍。

继续按铁路列车这个思路思考，我想知道如果有两辆列车会怎么样？我觉得应该能提高吞吐量，同时还能降低延迟。每个车站都会有它自己的列车。当一辆列车在第一个车站装货时，第二辆列车会在第二个车站卸货，反之亦然。

图 7: 使用两辆列车的单一生产者和单一消费者railway

下面是吞吐量的结果：

结果是惊人的；比单辆列车的结果快了1.4倍多。列车容量为一时，延迟从192.6纳秒降低到133.5纳秒；这显然是一个令人鼓舞的迹象。

因此我的实验还没结束。列车容量为2048的两个线程传递消息的延迟为2,178.4 纳秒，这太高了。我在想如何降低它，创建一个有很多辆列车 的例子：

图 8: 使用多辆列车的单一生产者和单一消费者railway 

我还把列车容量降到了1个long值，开始玩起了列车数量。下面是测试结果：

用32,768 列车在线程间发送一个long值的延迟降低到了13.9 纳秒。通过调整列车数量和列车容量，当延时不那么高，吞吐量不那么低时，吞吐量和延时就达到了最佳平衡。

对于单一生产者和单一消费者(spsc)而言，这些数值很棒；但我们怎么让它在有多个生产者和消费者时也能生效呢？答案很简单，添加更多的车站！

图 9：一个生产者和两个消费者的railway

每个线程都等着下一趟列车，装货/卸货，然后把列车送到下一站。在生产者往列车上装货时，消费者在从列车上卸货。列车周而复始地从一个车站转到另一个车站。

为了测试单一生产者/多消费者(spmc) 的情况，我创建了一个有8个车站的railway测试。 一个车站属于一个生产者，而另外7个车站属于消费者。结果是：

列车数量 = 256 ，列车容量 = 32:

 ops/sec = 116,604,397     延迟（纳秒） = 274.4

列车数量= 32，列车容量= 256:

 ops/sec = 432,055,469     延迟（纳秒） = 592.5

如你所见，即便有8个工作线程，测试给出的结果也相当好– 32辆容量为256个long的列车吞吐量为432,055,469 ops/sec。在测试期间，所有cpu内核的负载都是100%。

图 10：在测试有8个车站的railway 期间的cpu 使用情况

在玩这个railway算法时，我几乎忘了我最初的目标：提升多生产者/单消费者情况下的性能。

图 11：三个生产者和一个消费者的 railway 

我创建了3个生产者和1个消费者的新测试。每辆列车一站一站地转圈，而每个生产者只给每辆车装1/3容量的货。消费者取出每辆车上三个生产者给出的全部三项货物。性能测试给出的平均结果如下所示：

 ops/sec = 162,597,109  列车/秒 = 54,199,036     延迟（纳秒） = 18.5

结果相当棒。生产者和消费者工作的速度超过了160m ops/sec。

为了填补差异，下面给出相同情况下的disruptor结果- 3个生产者和1个消费者：

run 0, disruptor=11,467,889 ops/sec run 1, disruptor=11,280,315 ops/sec run 2, disruptor=11,286,681 ops/sec run 3, disruptor=11,254,924 ops/sec

下面是另一个批量消息的disruptor 3p:1c 测试 (10 条消息每批)：

run 0, disruptor=116,009,280 ops/sec run 1, disruptor=128,205,128 ops/sec run 2, disruptor=101,317,122 ops/sec run 3, disruptor=98,716,683 ops/sec;

最后是用带linkedblockingqueue 实现的disruptor 在3p:1c场景下的测试结果：

run 0, blockingqueue=4,546,281 ops/sec run 1, blockingqueue=4,508,769 ops/sec run 2, blockingqueue=4,101,386 ops/sec run 3, blockingqueue=4,124,561 ops/sec

如你所见，railway方式的平均吞吐量是162,597,109 ops/sec，而disruptor在同样的情况下的最好结果只有128,205,128 ops/sec。至于 linkedblockingqueue，最好的结果只有4,546,281 ops/sec。

railway算法为事件批处理提供了一种可以显著增加吞吐量的简易办法。通过调整列车容量或列车数量，很容易达成想要的吞吐量/延迟。

另外， 当同一个线程可以用来消费消息，处理它们并向环中返回结果时，通过混合生产者和消费者，railway也能用来处理复杂的情况：

图 12: 混合生产者和消费者的railway

最后，我会提供一个经过优化的超高吞吐量 单生产者/单消费者测试：

图 13：单个生产者和单个消费者的railway

它的平均结果为：吞吐量超过每秒15亿 (1,569,884,271)次操作，延迟为1.3 微秒。如你所见，本文开头描述的那个规模相同的单线程测试的结果是每秒2,681,850,373。

你自己想想结论是什么吧。

我希望将来再写一篇文章，阐明如何用queue和 blockingqueue接口支持railway算法，用来处理不同的生产者和消费者组合。敬请关注。