`

Java里快如闪电的线程间通讯

    博客分类:
  • JAVA
 
阅读更多

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

只用普通的老式Java接口实现POJO的通讯。

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

当actor跨越不同JVM实例(在同一台机器上,或分布在网络上的不同机器上)的进程边界时,Akka框架很善于处理进程间的通讯。

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

我花了几天时间,用动态代理,阻塞队列和缓存线程池创建了一个解决方案。

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


图一: 框架的高层次架构

 

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

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

 

LinkedBlockingQueue

Disruptor

Run 0

4,550,625 ops/sec

11,487,650 ops/sec

Run 1

4,651,162 ops/sec

11,049,723 ops/sec

Run 2

4,404,316 ops/sec

11,142,061 ops/sec

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

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

我创建了 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);    
  }    
 }    
}

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

容量

吞吐量: ops/sec

延迟: ns

1

5,190,883

192.6

2

10,282,820

194.5

32

104,878,614

305.1

256

344,614,640

742. 9

2048

608,112,493

3,367.8

32768

767,028,751

42,720.7

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

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

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

下面是吞吐量的结果:

容量

吞吐量: ops/sec

延时: ns

1

7,492,684

133.5

2

14,754,786

135.5

32

174,227,656

183.7

256

613,555,475

417.2

2048

940,144,900

2,178.4

32768

797,806,764

41,072.6

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

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

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

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

列车数量

吞吐量: ops/sec

延迟: ns

2

10,917,951

91.6

32

31,233,310

32.0

256

42,791,962

23.4

1024

53,220,057

18.8

32768

71,812,166

13.9

用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。

 
 
分享到:
评论

相关推荐

Global site tag (gtag.js) - Google Analytics