解决的问题
当我们有多个消息的生产者线程,一个消费者线程时,他们之间如何进行高并发、线程安全的协调?
很简单,用一个队列。
当我们有多个消息的生产者线程,多个消费者线程,并且每一条消息需要被所有的消费者都消费一次(这就不是一般队列,只消费一次的语义了),该怎么做?
这时仍然需要一个队列。但是:
1. 每个消费者需要自己维护一个指针,知道自己消费了队列中多少数据。这样同一条消息,可以被多个人独立消费。
2. 队列需要一个全局指针,指向最后一条被所有生产者加入的消息。消费者在消费数据时,不能消费到这个全局指针之后的位置——因为这个全局指针,已经是代表队列中最后一条可以被消费的消息了。
3. 需要协调所有消费者,在消费完所有队列中的消息后,阻塞等待。
4. 如果消费者之间有依赖关系,即对同一条消息的消费顺序,在业务上有固定的要求,那么还需要处理谁先消费,谁后消费同一条消息的问题。
总而言之,如果有多个生产者,多个消费者,并且同一条消息要给到所有的消费者都去处理一下,需要做到以上4点。这是不容易的。
LMAX Disruptor,正是这种场景下,满足以上4点要求的单机跨线程消息传递、分发的开源、高性能实现。
关键概念
1. RingBuffer
应用需要传递的消息在Disrutpor中称为Event(事件)。
RingBuffer是Event的数组,实现了阻塞队列的语义:
如果RingBuffer满了,则生产者会阻塞等待。
如果RingBuffer空了,则消费者会阻塞等待。
2. Sequence
在上文中,我提到“每个消费者需要自己维护一个指针”。这里的指针就是一个单调递增长整数(及其基于CAS的加法、获取操作),称为Sequence。
除了每个消费者需要维护一个指针外,RingBuffer自身也要维护一个全局指针(如上一节第2点所提到的),记录最后一条可以被消费的消息。这个全局指针就在下图红框中。
生产场景实现
生产者往RingBuffer中发送一条消息(RingBuffer.publish())时:
1. 生产者的私有sequence会+1
2. 检查生产者的私有sequence与RingBuffer中Event个数的关系。如果发现Event数组满了(下图红框中的判断),则阻塞(下图绿框中的等待)。
3. RingBuffer会在Event数组中(sequencer+1) % BUFFER_SIZE的地方,放入Event。这里的取模操作,就体现了Event数组用到最后,则回到头部继续放,所谓“Ring“ Buffer的轮循复用语义。
消费场景实现
消费者从RingBuffer循环队列中获取一条消息时:
1. 从消费者私有Sequence,可以知道它自己消费到了RingBuffer队列中的哪一条消息。
2. 从RingBuffer的全局指针Sequence,可以知道RingBuffer中最后一条没有被消费的消息在什么位置。
3. N = (RuingBuffer的全局指针Sequence - 消费者私有Sequence),就是当前消费者,还可以消费多少Event。
4. 如果以上差值N为0,说明当前消费者已经消费过RingBuffer中的所有消息了。那么当前消费者会阻塞。等待生产者加入更多的消息:
以上代码中,红框中的availableSequence就是RingBuffer的全局指针Sequence。绿框中的sequence是当前消费者的私有sequence。
如果这个判断为true,说明RingBuffer中最新一条可以被消费的Event,已经被当前消费者消费过了。那么就会调用apployWaitMethod()阻塞,等待生产者产生更多的Event。
5. 如果RingBuffer中,还有可以被当前消费者消费的Event,即N > 0,
那么消费者,会一口气获取所有可以被消费的N个Event。即下图中的while循环,直到N个Event都被消费才退出。这种一口气消费尽量多的Event,是高性能的体现。
从RingBuffer中每获取一个Event,都会回调绿框中的eventHandler——这是应用注册的Event处理方法,执行应用的Event消费业务逻辑。
最后,上图中的sequence.set(availableSequence),会把当前消费者的私有Sequence更新到RingBuffer的全局Sequence。表示RingBuffer中所有的Event都已经消费掉了。
高性能的实现细节
无锁
无锁就没有锁竞争。当生产者、消费者线程数很高时,意义重大。所以,
往大里说,每个消费者维护自己的Sequence,基本没有跨线程共享的状态。
往小里说,Sequence的加法是CAS实现的。
- 当生产者需要判断RingBuffer是否已满时,用CAS比较原先RingBuffer的Event个数,和假定放入新Event后Event的个数。
- 如果CAS返回false,说明在判断期间,别的生产者加入了新Event;或者别的消费者拿走了Event。那么当前判断无效,需要重新判断。这就是常见的 do { ... } while (false == CAS(oldVal, newVal))。——都是套路:)
对象的复用
JVM运行时,一怕创建大对象,二怕创建很多小对象。这都会导致JVM堆碎片化、对象元数据存储的额外开销大。这是高性能Java应用的噩梦。
为了解决第二点“很多小对象”,主流开源框架都会自己维护、复用对象池。LMAX Disruptor也不例外。
生产者不是创建新的Event对象,放入到RingBuffer中。而是从RingBuffer中取出一个已有的Event对象,更新它所指向的业务数据,来代表一个逻辑上的新Event。
所以LMAX Disruptor的生产者API,用起来有些麻烦——分为三步,一是下图绿框中取出一个已有的、已经被所有人消费过的Event对象,二是下图红框中更新这个Event对象所指向的业务数据,三是下图蓝框中标记这个Event对象为逻辑上的新Event。
、
相关推荐
Disruptor框架是由LMAX公司开发的一款高效的无锁内存队列。使用无锁的方式实现了一个环形队列。据官方描述,其性能要比BlockingQueue至少高一个数量级。根据GitHub上的最新版本源码打出的包,希望对大家有帮助。
LMAX Disruptor 最新版本 源码+API+驱动包
LMAX是一种新型零售金融交 易平台,它能够以很低的延迟产生大量交易。这个系统是建立在JVM平台上,其核心是一个业务逻辑处理 器,它能够在一个线程里每秒处理6百万订单。业务逻辑处理器完全是运行在内存中,使用事件...
这是LMAX Disruptor移植到Go编程语言中的移植。 它保留了Disruptor的本质和精神,并利用了许多相同的抽象和概念,但没有维护相同的API。 在使用Go 1.13.1的MacBook Pro(英特尔酷睿i9-8950HK CPU @ 2.90GHz)上,它...
Error: java.lang.NoSuchMethodError: com.lmax.disruptor.dsl.Disruptor.<init>(Lcom/lmax/disruptor/EventFactory;ILjava/util/concurrent/ThreadFactory;Lcom/lmax/disruptor/dsl/ProducerType;Lcom/lmax/...
Disruptor是一个高效的线程间交换数据的基础组件,它使用栅栏(barrier)+序号(Sequencing)机制协调生产者与消费者,从而避免使用锁和CAS,同时还组合使用预分配内存机制、缓存行机制(cache line)、批处理效应...
特征 单一生产者 批量消费者 阻塞等待策略 旋转等待策略 多制片人 工人池 DSL 文献资料基准测试从生产者向消费者发送i32大小的消息的初步基准测试结果。 名称批量大小通量mpsc频道1 34.894 Melem /秒驱油剂纺丝1个38...
显然,为了实现这一目标,我们需要做一些特殊的事情,以通过我们的Java平台实现极低的延迟和高吞吐量。 性能测试表明,使用队列在系统各阶段之间传递数据会引入延迟,因此我们专注于优化该区域。 Disruptor是我们...
Dispatcher 允许您注册多个生产者以及将消费者添加到命名事件通道。 Go 的接口被大量使用,以允许扩展的最大灵活性。 每个事件都必须实现Channel() []byte和Data []byte 。 将字节数组用于这两者可以实现最大的灵活...
Disruptor是一个开源的Java框架,它被设计用于在生产者—消费者(producer-consumer problem,简称PCP)问题上获得尽量高的吞吐量(TPS)和尽量低的延迟。Disruptor是LMAX在线交易平台的关键组成部分,LMAX平台使用...
最大值 LMAX Disruptor使Rx Java最大化
match-trade超高效的交易所撮合引擎,采用伦敦外汇交易所LMAX开源的Disruptor框架,分布式内存存取,以及原子性操作。使用数据流的方式进行计算撮合序列,才用价格水平独立撮合逻辑,实现高效大数据撮合
卡夫卡消费者的破坏者演示如何在Kafka 0.9 Consumer上使用LMAX Disruptor 好处->一旦先前的使用者完全处理完消息,便可以使用序列屏障来提交消息。想象力是极限。如果环形缓冲区可以容纳在L3缓存中,则处理速度会更...
在此实现中,日志记录和复制步骤同时发生,并且都必须成功才能执行发布步骤,如下所示: replicate / \ / \ event -> post \ / \ / journal 这也是基于“事件源”的内存中存储的示例-请参阅Martin Fowler:
Martin Fowler在自己网站上写了一篇LMAX架构的文章,在文章中他...Disruptor是一个高性能的异步处理框架,或者可以认为是最快的消息框架(轻量的JMS),也可以认为是一个观察者模式的实现,或者事件监听模式的实现。
提供相似或更好的性能特征提供与java实现类似的接口当前实现仅支持单个生产者在其当前状态下,性能测试大致显示了对 java 实现的 x2 性能改进,如 one_to_one_sequenced_throughput_test 的输出所示Java Run 0, ...
实现java Disruptor v3.3.7中可用的所有功能。 建造 编译器 Clang 3.8或更高版本 GCC 5.0或更高版本 Microsoft Visual C ++ 2015或更高版本 Linux 必须在您的计算机上可用。 您可以使用自己喜欢的软件包管理器进行...
赠送jar包:disruptor-3.3.0.jar; 赠送原API文档:disruptor-3.3.0-javadoc.jar; 赠送源代码:disruptor-3.3.0-sources.jar; 赠送Maven依赖信息文件:disruptor-3.3.0.pom; 包含翻译后的API文档:disruptor-...
移相器 LMAX Disruptor 的 Clojure DSL安装将以下依赖项添加到您的project.clj文件中: ; ; Stable[userevents/phaser " 1.1.5 " ]; ; or for the brave, use the cutting edge[userevents/phaser " 1.1.6-SNAPSHOT ...
Disruptor.Net ... 为多生产者和多消费者添加了 Disruptor.RingQueueTest,由 Shines77。 并将项目更改为在 Visual Studio 2013 更新 2 中使用 .Net Framework 4.5。 11/7/2013 (v2.10.0): Java Disrupto