BT

您是否属于早期采用者或者创新人士?InfoQ正在努力为您设计更多新功能。了解更多

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

| 作者 Aliaksei Papou 关注 0 他的粉丝 ,译者 吴海星 关注 0 他的粉丝 发布于 2014年2月21日. 估计阅读时间: 4 分钟 | 智能化运维、Serverless、DevOps......2017年有哪些最新运维技术趋势?CNUTCon即将为你揭秘!

这个故事源自一个很简单的想法:创建一个对开发人员友好的、简单轻量的线程间通讯框架,完全不用锁、同步器、信号量、等待和通知,在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。

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

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

关于作者

Aliaksei PapouSpecific集团的首席软件工程师和架构师,那是一家位于奥地利维也纳的软件开发公司。Aliaksei有超过10年的小型和大型企业应用软件开发经验。他有一个坚定的信念:编写并发代码不应该这么难。

 

原文英文链接:Inter-thread communications in Java at the speed of light


感谢侯伯薇对本文的审校。

给InfoQ中文站投稿或者参与内容翻译工作,请邮件至editors@cn.infoq.com。也欢迎大家通过新浪微博(@InfoQ)或者腾讯微博(@InfoQ)关注我们,并与我们的编辑和其他读者朋友交流。

评价本文

专业度
风格

您好,朋友!

您需要 注册一个InfoQ账号 或者 才能进行评论。在您完成注册后还需要进行一些设置。

获得来自InfoQ的更多体验。

告诉我们您的想法

允许的HTML标签: a,b,br,blockquote,i,li,pre,u,ul,p

当有人回复此评论时请E-mail通知我

很巧妙的算法 by xiaoran wu

值得尝试应用的算法

用了8个核优化了半天还不如简单的单线程的效果,这就是结论吧? by wang xinkai

business logic 过于简单了,所以时间统统都浪费在overhead上了,所以单核单线程反而更快。
其实玩这些东西时跟JAVA语言本身的关系已经不大了,用C#玩多线程优化也是一样的道理。

Github有一个Reactor 和这个非常像 by yamato kira

也是Actor模式的 其中的Dispatcher也有使用到Disruptor

总的来看这是个令牌环模型 by 林 大海

早期同轴电缆网络用的通讯模型就是令牌环模型,与之对比的是:网卡替换为车站,令牌替换为车厢。最后的优化方案是采用了多令牌策略。总的来说挺Nice的~~~

收藏 by 冯 光头

收藏

想知道这个是用什么画得图 by lee jacky

开始看到的线程模型非常简单,但描述的很清楚,看到后面的方案演进,

可能实际应用中的问题 by 王 超

这个算法思路不错,有一些可以借鉴的地方,但这个算法在执行过程中CPU利用率很高,可能会导致CPU长时间过热。

Re: 可能实际应用中的问题 by 王 超

其实可以更深入的思考一下,高铁、地铁其实都不是24小时运营的,当旅客少或是没有旅客时,可以考虑暂时停运而减少不必要的CPU开销。当出现旅客或旅客达到一定量时,再启动运行。

类似于node中的事件调度器 by Yu Yole

类似于node中的事件调度器

mark by anjin yu

mark一下

原作者的测试代码在这里 by shen zixian

业务太简单了,不实际 by yan xilang

因为业务非常简单,所以CPU都可以用在循环等待结果上。如果业务复杂,那个不停循环的线程会不会占用了本该用来处理业务所需的CPU资源呢?

Re: 总的来看这是个令牌环模型 by jiang jianbo

我也是想到了令牌环,直到看这篇文章之前,我还对令牌环的威力没有感觉呢。非常好的文章,感谢作者和译者,谢谢!

学习了,列车的数量要是能根据consumer的数量来浮动就好了 by kk xk

学习了,列车的数量要是能根据consumer的数量来浮动就好了

原文的评论也值得一读 by Xikun Yang

Re: 业务太简单了,不实际 by Zhefu Zhou

任何复杂的业务逻辑都不应该出现在循环线程上,这一点几乎是所有框架的共识,Disruptor的RingBuffer、Netty的bossGroup都是如此。令牌环只管分发。

Re: 可能实际应用中的问题 by Zhefu Zhou

高性能框架的目的就是充分利用现有的硬件资源,让CPU长时间全速工作而非等待,如果CPU过热,那么说明架构的目的达到了。

除非系统明确定义了宕机时间,否则不可能“停运”。无论是令牌环还是RingBuffer,分发线程(生产者)必定是要永远循环下去的,而负荷低峰期所减少的,是活动的worker(消费者)线程。

高效,但是感觉消息不是顺序被取走的。 by 张 恭慆

现实中,消费者一般要耗费时间去处理业务逻辑,而且很多是需要顺序处理的,顺序乱了可能会造成数据的混乱,这个时候,这种方法就不行了。不过,这个思路确实从测试结果看,效率很高。非常值得学习。不知道我说的对不对。欢迎板砖。

github上的测试代码。。。 by 李 清风

根本没办法拿来用。。。

算法不实用吧 by 胡 li

用线程去循环消耗cpu 这其实就是自旋锁的一个实现而已 如果其他业务复杂的情况下表现应该不是很好

代码有重大缺陷,不可使用。 by William Zhi

注意它的消费者模型,用了一个trainIndex++来等车。好列车来了,处理,发送,然后等下一班列车。下一班车的号码是啥?上一班车的序号加一。那如果下一班车因为发送者卡壳,永远都不来了呢?其他生产者发出的消息就永远都得不到处理。这个模型在实际生产环境中会出大问题。

允许的HTML标签: a,b,br,blockquote,i,li,pre,u,ul,p

当有人回复此评论时请E-mail通知我

允许的HTML标签: a,b,br,blockquote,i,li,pre,u,ul,p

当有人回复此评论时请E-mail通知我

21 讨论

登陆InfoQ,与你最关心的话题互动。


找回密码....

Follow

关注你最喜爱的话题和作者

快速浏览网站内你所感兴趣话题的精选内容。

Like

内容自由定制

选择想要阅读的主题和喜爱的作者定制自己的新闻源。

Notifications

获取更新

设置通知机制以获取内容更新对您而言是否重要

BT