thread_Disruptor

时间 2019-12-12

标签 thread disruptor 繁體版

原文原文链接

转自知乎 https://zhuanlan.zhihu.com/p/21355046html

order从client端传入，decode后进行matching，一旦存在可成交的价格，就要publish到time series，而且把trade存到local的database里。如何handle这么大数量的数据？java

这并非一个新生的问题。一个常常想到的模型是producer consumer model。git

当系统的处理速度比不上导入数据的速度时，能够增长一个queue（buffer）暂存数据，等待consumer处理。数据在queue中被执行的顺序和交易策略有关。github

除了handle大量数据，电子商务对数据延迟的要求也很高。首先定义一下什么是数据的延迟。多线程

数据的延迟包括数据的处理时间和数据的移动时间。其中第二部分在编写代码时常被忽略，但在实际生产过程当中常占有很大比重。app

Blocking queue有两种，array-based和linked list based。array-based相对更优，这里咱们先对它进行一些分析。性能

Producer和consumer处理速度每每是不一样的，这样容易造成两种状况：一种是producer速度快，queue易全满，另外一种是consumer速度快，queue易全空。google

Blocking queue的缺点主要有两个：一个是producer只能从head放数据，producer之间会竞争head指针，存在写竞争。consumer之间会竞争tail指针，它们之间也存在写竞争。而且不少状况下，queue是处于全空状态，head/tail指针指向同一个entry，producer和consumer之间也存在写竞争。所以须要lock来实现synchronization。另外一个缺点是heal/tail指针的false sharing。atom

在进行下面讲解前先下两个结论，理由后续会涉及。spa

如今的计算机构架每每是有CPU，memory，它们之间有多层的cache。这种构架产生的缘由在于CPU速度远高于memory速度。

为了解决速度间不一样步，可使用cache。Cache是对历史数据的保存。若是数据以前没有读取，不存在于本层cache，则须要从上层cache里获取，存到本层cache里；若以前有读取，本层cache存有数据历史信息。有了cache，数据的移动路径变短。同时，写操做相对费时，CPU会在较优的时间进行写操做。

具体cache如何工做，见下图：

Memory的最小单位叫block，cache的最小单位叫cache line。Memory到cache存在多对一的mapping。图中用相同颜色表示它们之间的mapping。举例说明，若是有一个integer array，里面存有10个数据。它们一一映射到cache里。

但每每cache的尺寸小于memory，19，20没法写入。对于多对一的这种mapping，存在对原有数据的eviction，19，20写在原来11，12处。

Memory的block每每大于一个integer。在读取A[0]时，实际上也把A[1]读取进去。因此在实际读取A[1]时，它已经存在于cache里。这是一种spatial cache locality。

若是A[9]变成35，只须要对cache里的数据进行更改。

若是使用多个核，这种方法会出现问题。好比第二个核里的数据仍是原来的20。

多核中对数据修改时，若是数据存在于多个核的cache里，要将其余核里数据设为invalid。

下面介绍false sharing。假设有两个integer a = 13和b = 14。若是第一个核在访问a，第二个核在访问b。

Core 1在访问a时让core 2中对应数据invalid，core 2修改时发现invalid，从新读取数据。可是core 2在读写时又把core 1对应数据invalid。Blocking queue里由于head/tail指针常是同一个，而producer和consumer在不一样的core上运行，常会发生上述的false sharing，加大了数据移动的时间。

为何使用lock会形成不少数据的移动？

如下图为例：

Core 1里有thread 1在运行，当遇到lock后，thread 1 sleep，core 1里运行thread 2。对于thread 2，core 1里cache的数据都是无用的。

Thread 2从新加载数据运行。当thread 1醒来时，只能在core 2上运行，从新加载数据。因此当有lock的时候，出现了不少的cache miss，增长了数据移动的时间。

总结一下，blocking queue很慢的缘由在于：写竞争形成的thread arbitrage以及false sharing致使的不少memory access。

Design of Disruptor

在设计Disruptor时要避免写竞争，让数据更久的留在cache里。

设计原则有：只有一个consumer，避免使用lock等等。

Disruptor的核心是一个circular array，有个cursor，里面有sequence number，数据类型是long。若是不考虑consumer，只有一个producer在写，就是不停的往entry里写东西，而后增长cursor上的sequence number。为了不cursor里的sequence number和其余variable造做false sharing，disruptor定义了7个long型，并无给它们赋值，而后再定义cursor。这样cursor就不会和其余variable同时出如今一个cache line里。

若是producer在写的过程当中，超出了原来array的长度，就不停地overwrite原来的entry，增长cursor里的sequence number。bucket里的entry都是pre-allocated，避免每次都new一个object。由于disruptor是用java写的，这样能够避免garbage collection。producer写的过程是two phase commit。

若是加入了consumer，以下图：

若是consumer当前访问的sequence number为5，producer当前访问到18。那么consumer能够一路访问到18，producer往前写不能超过5。

若是有多个consumer：

在disruptor里不一样consumer之间没有contention。如上图中consumer 1能够从5读到18，consumer 2能够不用管consumer 1的存在，也一路读到18，consumer之间能够忽略对方的存在。

Consumer每次在访问时须要先检查sequence number是否available，若是不available，会有多种策略。latency最高的一种是盲等。producer在写的时候，须要检查最低的sequence number在哪儿。这里不须要lock的缘由是sequence number是递增的。producer不须要赶在最低sequence number前面，于是没有write contention。此外，disruptor使用memory barrier通知数据的更新。

Memory barrier

CPU认为逻辑上没有冲突的instruction能够reorder。写操做须要花不少时间，能够在schedule pipeline比较方便的时候把instruction插进去。好比core 1须要写a，b，c，d。由于这四个variable之间没有关系，它们的顺序也是能够打乱的。在disruptor中并不直接把它们写入cache中，而是写入core和cache直接的一个store buffer里，在store buffer里四个variable是reorder的。

单线程下没有任何问题，可是多线程时，core 2角度来看，c先被写，而后是d，a，b。在disruptor里producer最后update cursor里的sequence number，告诉你们这个entry已经ready，全部的consumer能够读它。可是若是写entry的顺序和写sequence number的顺序不一致，会形成一种现象：sequence number的写已经完成，consumber能够去读对应数据，可是对应的entry的写尚未ready。