【优化技术专题】「线程间的高性能消息框架」再次细节领略Disruptor的底层原理和优势分析

简介: 【优化技术专题】「线程间的高性能消息框架」再次细节领略Disruptor的底层原理和优势分析

Disruptor原理


首先Disruptor是为了解决高并发缓存的队列,为线程间通讯提供高效的性能,它是如何做到无阻塞、多生产、多消费的?

image.png


上图简单的画了一下构建Disruptor的各个参数以及 ringBuffer 的构造,下面简单的说一下。



生产者需要组件


生产者,产生消息,并将消息发布到RingBuffer内存队列中。


  • Event模型:从生产者传递给消费者的数据单位,完全由用户定义其类型。
@Data
public class SampleEvent {
    private Long id。
    private String sampleDataStr。
}
复制代码
  • EventFactory:创建事件(任务)的工厂类。(这里任务会创建好,保存在内存中,可以看做是一个空任务)。
public class SampleEventFactory implements EventFactory<SampleEvent> {
    @Override
    public SampleEvent newInstance() {
        // 实例化数据(建好空数据,等待后面初始化)
        return new SampleEvent()。
    }
}
复制代码


  • RingBuffer:环形缓冲区通常被认为是Disruptor的主要实现,当前版本即3.0版本之后,RingBuffer仅负责存储和更新通过Disruptor的数据(Event)。
  • ringBufferSize:容器的长度。( Disruptor 的核心容器是 ringBuffer,环转数组,有限长度)。
  • ProductType:生产者类型:单生产者、多生产者。
  • Sequencer:Sequencer是Disruptor的核心API。该接口的2个实现类(SingleProducer,MultiProducer)实现了所有并发算法,用于在生产者和消费者之间快速,正确地传递数据。
  • WaitStrategy:等待策略。(当队列里的数据都被消费完之后,消费者和生产者之间的等待策略),等待策略确定消费者如何等待生产者将事件放入Disruptor。


  • RingBuffer:存放数据的容器。


@Data
@AllArgsConstructor
public class SampleEventProducer {
    private RingBuffer<OrderEvent> ringBuffer。
    public void sendData(long id) {
        //获取下一个可用序号
        long sequence = ringBuffer.next()。
        try {
            //获取一个空对象(没有填充值)
            SampleEvent sampleEent = ringBuffer.get(sequence)。
        }finally {
            //提交
            ringBuffer.publish(sequence)。
        }
    }
}
复制代码




消费者需要组件


  • Executor:消费者线程池,执行任务的线程。(每一个消费者都需要从线程池里获得线程去消费任务)。


  • EventProcessor:用于处理来自Disruptor的事件的主事件循环,并具有消费者序列的所有权。有一个名为 BatchEventProcessor表示,它包含事件循环的有效实现,并将回调到使用的提供的EventHandler接口实现。


  • EventHandler:事件处理器,由用户实现并代表Disruptor的使用者的接口,用户客户端实现消息的处理机制,由客户端具体实现。


public class SampleEventHandler implements EventHandler<SampleEvent> {
    /**
     * 事件驱动监听--消费者消费的主体
     */
    @Override
    public void onEvent(SampleEvent  event, long sequence, boolean endOfBatch) throws Exception {
        System.out.println(event.getSampleDataStr() + "  " +Thread.currentThread().getName())。
    }
}
复制代码



算法核心Sequence序号


  • Sequence:Disruptor使用Sequences作为识别特定组件所在位置的方法。
  • 每个消费者(EventProcessor)都像Disruptor本身一样维护一个Sequence。大多数并发代码依赖于这些Sequence值的变化或者叫移动,因此Sequence支持AtomicLong的许多当前功能。
  • 事实上,唯一真正的区别是Sequence包含额外的功能,以防止序列和其他值之间的错误共享。
  • Sequence Barrier:序列屏障由Sequencer产生,包含对Sequencer中主要发布的sequence和任何依赖性消费者的序列的引用。它包含确定是否有任何可供消费者处理的事件的逻辑。



Disruptor的优点:


  1. 多线程之间没有竞争即没有锁。


  1. 所有访问者都记录自己的序号的实现方式,允许多个生产者与多个消费者共享相同的数据结构。


  1. 每个对象中都能跟踪序列号(ring buffer, claim strategy,生产者和消费者),加上神奇的缓存行填充,就意味着没有伪共享和非预期的竞争。


下面再简单介绍下RingBuffer核心实现,来看看队列的实现细节。

image.png


其为环形队列,有点像一致性Hash算法中的闭环,但完全不一样。


底层的话是一个固定大小的数组结构,相比于队列来说,其只有一个下标指针cursor,如果槽的个数是2的N次方更有利于基于二进制的计算机进行计算。如果看过HashMap源码应该知道,HashMap定位元素槽时使用了一种巧妙的方式,hash&(length-1)。

image.png



RingBuffer同样是相同的计算方式,sequence&(length-1),当然你可以进行取模操作。


  • 取模操作在寄存器中的计算,需要多次的迭代加操作进行的,所以相对于计算速度来说,对于计算机进行位运算效率绝对是高于取模操作的,尤其是对于高并发状况下的计算,能够节省很多单位cpu开销


一般实现线性存储有两种实现方式:


  • 一种是基于连续内存分配的HashTable
  • 一种是基于随机内存分配的迭代指针。


为什么RingBuffer选用数组作为存储结构,而不选用链表存储?



缓存或者程序的局部性原理


  • (Good)数组内存属是连续分配内存的预读策略,也就是内存加载时,会将部分连续内存地址预先加载到高速缓存中,即认为你可能会使用,上面我们分析了操作系统中的cpu操作数据的流程,可以看出这种设计是为了不用反复从内存中加载。
  • (Bad)链表的内存分配是碎片化的所以其存储地址不是连续的,导致每次都会cpu都会重新计算下一个链表位置的地址,并从内存中加载相关的数据,数据量小的情况下并不能看出性能的优劣,但是当数据量大的情况下,这种极小的消耗,会对整体的运行效率产生影响。

因为RingBuffer不会涉及到存储地址的修改和维护,就因为选用数组就对性能产生了有利和积极的影响。



伪共享


内存以高速缓存行的形式存储在高速缓存系统中。高速缓存行是2的N次方个连续字节,其大小通常为32-256,最常见的缓存行大小为64字节


伪共享是一个术语,适用于线程在修改共享同一缓存行的独立变量时无意中影响彼此的性能。在高速缓存行上写入争用是实现SMP系统中并行执行线程的可伸缩性的最大限制因素。(出自百度定义!)


  • 首先我们知道对于锁来说是关中断实现,锁定bus消息总线实现,而对于共享内存,计算机使用的是缓存行,共享变量的多个线程,共享相同的缓存行。
  • 实现线程数量的线性可伸缩性,我们必须确保没有两个线程写入同一个变量或缓存行。而当使用volatile的时候,我们读取直接共享变量从主内存或者叫共享内存中读取变量的值,其本质是使计算机缓存行失效。


在CPU核心A运行的线程想要更新变量X,而CPU核心B上的线程想要更新变量Y。


这两个热变量位于同一缓存行中。每个线程都将竞争缓存行的所有权,以便他们可以更新它。如果核心A获得所有权,那么MESI/MOSI缓存子系统将需要使核心B的相应缓存行无效。反之也是一样,极大地影响性能。如果竞争核心在不同的套接字上并且还必须跨越套接字互连,则缓存行问题将进一步加剧。


总结一下:如果多个线程操作不同的成员变量, 但是这些变量存储在同一个缓存行,如果有处理器更新了缓存行的数据并刷新到主存之后,根据缓存一致性原则,其他处理器将失效该缓存行(I状态)导致缓存未命中,需要重新去内存中读取最新数据,这就是伪共享问题。


  • 特别是不同的线程操作同一个缓存行,需要发出RFO(Request for Owner)信号锁定缓存行,保证写操作的原子性,此时其他线程不能操作这个缓存行,这将对效率有极大的影响。

为了避免避免经常执行写操作的变量因为在同一个缓存行而导致的伪共享问题,常用的解决方式就是缓存行填充,或者称为缓存行对齐。



缓存行填充的概念


当多个线程同时对共享的缓存行进行写操作的时候,因为缓存系统自身的缓存一致性原则,会引发伪共享问题,解决的常用办法是将共享变量根据缓存行大小进行补充对齐,使其加载到缓存时能够独享缓存行,避免与其他共享变量存储在同一个缓存行。



下面是缓存行实现,另外缓存行填充有一个前提同时分配的对象往往位于同一位置。

public long p1, p2, p3, p4, p5, p6, p7; // cache line padding
private volatile long cursor = INITIAL_CURSOR_VALUE;
public long p8, p9, p10, p11, p12, p13, p14; // cache line padding
复制代码



如果有不同的消费者往不同的字段写入,你需要确保各个字段间不会出现伪共享。

/**
 * 数组保存了VolatileLongPadding,其中数组中一个long类型保存数组长度,算上
 * 自身long类型value,需要再填充6个long类型,就能将数组中的对象填充满一个缓存行。
 * 注意:这里使用继承的方式实现缓存行对齐,因为Java编译器会优化无效的字段。
 */
class CacheLinePadding {
    // 如果不需要填充,只需要注释掉这段代码即可
    public volatile long p1, p2, p3, p4, p5, p6;
}
class CacheLinePaddingObject extends CacheLinePadding {
    //实际操作的值
    public volatile long value = 0L;
}
复制代码



RingBuffer实现上同样也是使用了缓存行填充,保证了数组中的数据没有伪共享的存在,RingBuffer除了一个long类型的cursor索引指针,p1->p7都为缓存行填充,一般来讲8个Long类型的字段,正好是64Byte,会填充一个缓存行,当然前提是你需要,因为毕竟还有你自己的数据信息字段。











相关文章
|
6天前
|
Go 调度 开发者
Go语言多线程的优势
【10月更文挑战第15天】
10 4
|
5天前
|
缓存 监控 Java
Java 线程池在高并发场景下有哪些优势和潜在问题?
Java 线程池在高并发场景下有哪些优势和潜在问题?
|
2月前
|
存储 缓存 Java
什么是线程池?从底层源码入手,深度解析线程池的工作原理
本文从底层源码入手,深度解析ThreadPoolExecutor底层源码,包括其核心字段、内部类和重要方法,另外对Executors工具类下的四种自带线程池源码进行解释。 阅读本文后,可以对线程池的工作原理、七大参数、生命周期、拒绝策略等内容拥有更深入的认识。
116 29
什么是线程池?从底层源码入手,深度解析线程池的工作原理
|
16天前
|
调度 Android开发 开发者
构建高效Android应用:探究Kotlin多线程优化策略
【10月更文挑战第11天】本文探讨了如何在Kotlin中实现高效的多线程方案,特别是在Android应用开发中。通过介绍Kotlin协程的基础知识、异步数据加载的实际案例,以及合理使用不同调度器的方法,帮助开发者提升应用性能和用户体验。
36 4
|
21天前
|
Java 数据库连接 数据库
不同业务使用同一个线程池发生死锁的技术探讨
【10月更文挑战第6天】在并发编程中,线程池是一种常用的优化手段,用于管理和复用线程资源,减少线程的创建和销毁开销。然而,当多个不同业务场景共用同一个线程池时,可能会引发一系列并发问题,其中死锁就是最为严重的一种。本文将深入探讨不同业务使用同一个线程池发生死锁的原因、影响及解决方案,旨在帮助开发者避免此类陷阱,提升系统的稳定性和可靠性。
38 5
|
26天前
|
缓存 负载均衡 Java
c++写高性能的任务流线程池(万字详解!)
本文介绍了一种高性能的任务流线程池设计,涵盖多种优化机制。首先介绍了Work Steal机制,通过任务偷窃提高资源利用率。接着讨论了优先级任务,使不同优先级的任务得到合理调度。然后提出了缓存机制,通过环形缓存队列提升程序负载能力。Local Thread机制则通过预先创建线程减少创建和销毁线程的开销。Lock Free机制进一步减少了锁的竞争。容量动态调整机制根据任务负载动态调整线程数量。批量处理机制提高了任务处理效率。此外,还介绍了负载均衡、避免等待、预测优化、减少复制等策略。最后,任务组的设计便于管理和复用多任务。整体设计旨在提升线程池的性能和稳定性。
66 5
|
19天前
|
网络协议 安全 Java
难懂,误点!将多线程技术应用于Python的异步事件循环
难懂,误点!将多线程技术应用于Python的异步事件循环
47 0
|
26天前
|
Java 编译器 程序员
【多线程】synchronized原理
【多线程】synchronized原理
45 0
|
26天前
|
Java 应用服务中间件 API
nginx线程池原理
nginx线程池原理
24 0
|
2月前
|
存储 消息中间件 NoSQL
Redis的单线程设计之谜:高性能与简洁并存
Redis的单线程设计之谜:高性能与简洁并存
41 0