首页 > 其他分享 >Disruptor-简单使用

Disruptor-简单使用

时间:2023-04-10 15:46:06浏览次数:47  
标签:Disruptor 简单 disruptor LongEvent value 使用 new Event

前言

Disruptor是一个高性能的无锁并发框架,其主要应用场景是在高并发、低延迟的系统中,如金融领域的交易系统,游戏服务器等。其优点就是非常快,号称能支撑每秒600万订单。需要注意的是,Disruptor是单机框架,对标JDK中的Queue,而非可用于分布式系统的MQ

本文基于Disruptor v3.4.*版本

Demo

既然是简单使用,这阶段只需要关注:

  • 生产者
  • 消费者:EventHandler
  • 消息的传递:消息的载体Event

简单例子

首先,我们定义消息的载体Event,生产者向消费者传递的消息通过Event承载

class LongEvent {
    private long value;

    public void set(long value) {
        this.value = value;
    }
    @Override
    public String toString() {
        return "LongEvent{" + "value=" + value + '}';
    }
}

然后定义Event生产工厂,这用于初始化Event

EventFactory<LongEvent> factory = new EventFactory<LongEvent>() {
    @Override
    public LongEvent newInstance() {
        return new LongEvent();
    }
};

接下来就可以构建Disruptor了,以下是完整代码

// 消息载体(event)
static class LongEvent {
    private long value;

    public void set(long value) {
        this.value = value;
    }
    @Override
    public String toString() {
        return "LongEvent{" + "value=" + value + '}';
    }
}

// 发布消息的转换器
public static void translate(LongEvent event, long sequence, ByteBuffer buffer)
{
    event.set(buffer.getLong(0));
}

public static void main(String[] args) throws Exception {

    // event生产工厂,初始化RingBuffer的时候使用
    EventFactory<LongEvent> factory = new EventFactory<LongEvent>() {
        @Override
        public LongEvent newInstance() {
            return new LongEvent();
        }
    };

    // 指定RingBuffer的大小(必须是2的n次方)
    int bufferSize = 1024;

    // 构造Disruptor(默认使用多生产者模式、BlockingWaitStrategy阻塞策略)
    Disruptor<LongEvent> disruptor = new Disruptor<>(LongEvent::new, bufferSize, DaemonThreadFactory.INSTANCE);
    //  Disruptor<LongEvent> disruptor = new Disruptor<>(factory, bufferSize, DaemonThreadFactory.INSTANCE, ProducerType.MULTI, new BlockingWaitStrategy());
    // 设置消费者
    EventHandler<LongEvent> handler = (event, sequence, endOfBatch) -> {
        System.out.println("Event: " + event);
    };
    disruptor.handleEventsWith(handler);

    // 启动disruptor,启动所有需要运行的线程
    disruptor.start();

    RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
    ByteBuffer bb = ByteBuffer.allocate(8);
    for (long i = 0; i < 100; i++) {
        bb.putLong(i);
        // 发布事件
        ringBuffer.publishEvent(LongEventMain::translate, bb);
    }
}

消费者组合(多使用场景)

Disruptor不仅可以当高性能的队列使用,还支持消费者的串行、并行消费等

以下只展示关键代码(设置消费者),其余部分参考上一节的简单demo

  1. 单链串行

    Untitled

    disruptor.handleEventsWith(handlerA).then(handlerB);
    
  2. 并行

    Untitled

    disruptor.handleEventsWith(handlerA, handlerB);
    
  3. 链内串行,多链并行

    Untitled

    disruptor.handleEventsWith(handlerA).then(handlerC);
    disruptor.handleEventsWith(handlerB).then(handlerD);
    
  4. 菱形(C、D都执行完才到E)

    Untitled

    disruptor.handleEventsWith(handlerA).then(handlerC);
    disruptor.handleEventsWith(handlerB).then(handlerD);
    disruptor.after(handlerC, handlerD).then(handlerE);
    
    
  5. 分组(AB都执行完才到CD)

    Untitled

    disruptor.handleEventsWith(handlerA, handlerB).then(handlerC, handlerD);
    
  6. 分组不重复消费

    组内竞争,组外串行:每个消息在每个分组中只有一个消费者能消费成功,如果就是分组A中只有HandlerA2能得到数据,分组B中只有HandlerB1获得

    Untitled

    // 注意:此处的handler实现的是WorkHandler接口
    disruptor.handleEventsWithWorkerPool(handlerA1, handlerA2, handlerA3)
                    .then(handlerB1, handlerB2, handlerB3);
    
  7. 分组不重复消费(菱形)

    Untitled

    // handlerA、handlerB实现WorkHandler接口
    // handlerC 实现EventHandler或WorkHandler接口均可
    disruptor.handleEventsWithWorkerPool(handlerA1, handlerA2, handlerA3)
                    .then(handlerB1, handlerB2, handlerB3)
                    .then(handlerC);
    

    等待策略

    消费者速度比生产者快时,需要等待。因此就有了不同的等待策略以适应不同场景

    • BlockingWaitStrategy

      默认策略。使用锁和 Condition 的等待、唤醒机制。速度慢,但节省CPU资源并且在不同部署环境中能提供更加一致的性能表现。

    • YieldingWaitStrategy

      二段式,一阶段自旋100次,二阶段执行Thread.yield,需要低延迟的场景可使用此策略

    • SleepingWaitStrategy

      三段式,一阶段自旋,二阶段执行Thread.yield,三阶段睡眠

    • BusySpinWaitStrategy

      性能最高的策略,与 YieldingWaitStrategy 一样在低延迟场景使用,但是此策略要求消费者数量低于 CPU 逻辑内核总数

    其他小技巧

    1. 清除消息载体 Event 中的数据

      如果 Event 中存在大对象,应该在消费者链的末尾,添加一个清除数据的消费者,以帮助jvm垃圾回收。demo中的 LongEvent 是 private long value; 所以没必要添加。

总结

本文介绍了 Disruptor 的简单使用,以及复杂场景下消费者的配置。下篇开坑 Disruptor 源码解析。


参考资料

Disruptor官方文档

标签:Disruptor,简单,disruptor,LongEvent,value,使用,new,Event
From: https://www.cnblogs.com/konghuanxi/p/17303118.html

相关文章

  • 【解决】Windows10如何限制CPU最大使用率?
    【问题】Windows10如何限制CPU最大使用率?【解决】:进入控制面板→电源选项→更改计划设置→更改高级电源设置→处理器电源管理,最大处理器状态。设置需要的百分比即可。  ......
  • DOSbox的安装及其运行和基本命令的使用(内附下载链接)
    下载工具AsmTools(内含4个文件)1个安装文件(DOSBox0.74-Win32-installer),3个调试工具(debug、LINK、MASM)链接:https://pan.baidu.com/s/12HX_hHye8upcCO9Wwm6Qtg提取码:1234 任选一个分区盘,如E盘在根目录下建立一个英文文件夹命名为“debug”,将3个调试工具(debug、LINK、MASM)放入......
  • Windows 下安装 MySQL 8/7(使用命令行)
    下载Mysql的zip安装包下载地址:https://mirrors.aliyun.com/mysql/MySQL-8.0/https://mirrors.163.com/mysql/Downloads/MySQL-8.0/https://mirrors.aliyun.com/mysql/MySQL-5.7/https://mirrors.163.com/mysql/Downloads/MySQL-5.7/创建my.ini配置文件文件解压,进入mysql......
  • clayui clayui使用配置
                  在使用CLAYUI前需要对您的工程做如下设置:DebugMultithreadedDLL或者MultithreadedDLL                      VC6下具体设置为:Project-Settings-C/C++,Category选择CodeGeneration,                   ......
  • 软件测试|超好用超简单的Python GUI库——tkinter(四)
    前言之前我们介绍了label控件,本篇文章我们将介绍button控件。Button控件是Tkinter中常用的窗口部件之一,同时也是实现程序与用户交互的主要控件。通过用户点击按钮的行为来执行回调函数,是Button控件的主要功用。首先自定义一个函数或者方法,然后将函数与按钮关联起来,最后,当用户......
  • clayui实用系列(一):在C#里切换窗口时使用各种3D/2D眩酷特效,多达13种特效。
       预览:         下载     这次clayui给大家带来了比较实用的东西,因为时间比较仓促,就先放出C#版,目前只有13种特效,因为做这个本身也比较麻烦,所以想先看看大家的反应吧,如果大家觉得不好用,或者不够眩的话,就不继续下去了。当然,如果大家对这个比较喜欢的话,会......
  • 软件测试|超好用超简单的Python GUI库——tkinter(五)
    前言在之前,我们介绍了tkinter的button控件,label控件,今天我们介绍一下entry控件,entry控件我们可以理解为界面的内容输入框,实现GUI界面与用户的信息交互,最典型的场景就是我们在登录时需要输入的账号密码。Entry控件使用起来非常简单,下面对该控件做简单的介绍。基本语法格式如下:tk_en......
  • 第9章 使用MVC为移动和客户端应用程序创建Web API(ASP.NET Core in Action, 2nd Editio
    本章包括创建WebAPI控制器以向客户端返回JSON使用属性路由自定义URL使用内容协商生成响应使用[ApiController]属性应用通用约定在前五章中,您已经完成了服务器端渲染ASP.NETCore应用程序的每一层,使用RazorPages将HTML渲染到浏览器。在本章中,您将看到对ASP.NETCore应用程......
  • 使用ChatGPT写一个GDALDatasetRasterIOEx功能增强函数
    GDALDatasetRasterIOEx函数是GDAL用来读取栅格数据文件像素数据的函数,默认是不支持参数传入的读取窗口范围超出栅格的实际范围的,但是很多时候,需要按照地理范围读取,就会有超出栅格范围的情况。之前我常用的做法,就是创建一个基于Mem的VRT(虚拟栅格),VRT的地理范围取自己想要读取的......
  • 第8章 使用标记帮助工具构建表单(ASP.NET Core in Action, 2nd Edition)
    本章包括使用TagHelpers轻松构建表单使用锚标记帮助程序生成URL使用TagHelpers为Razor添加功能在第7章中,您了解了Razor模板以及如何使用它们为应用程序生成视图。通过混合HTML和C#,您可以创建动态应用程序,根据请求、登录用户或您可以访问的任何其他数据显示不同的数据。显......