百度360必应搜狗淘宝本站头条
nodejsbufferorg.jsoupsetnocountoncv2.bitwise_nottime.strptime
当前位置：网站首页 > 编程字典 > 正文

Disruptor的简单介绍与应用

toyiye 2024-06-27 00:38 18 浏览 0 评论

前言

最近工作比较忙，在工作项目中，看了很多人都自己实现了一套数据任务处理机制，个人感觉有点乱，且也方便他人的后续维护，所以想到了一种数据处理模式，即生产者、缓冲队列、消费者的模式来统一大家的实现逻辑。

下面时是对Disruptor基本使用的演示。使用中需要引入依赖

<dependency>
  <groupId>com.lmax</groupId>
  <artifactId>disruptor</artifactId>
  <version>3.4.2</version>
</dependency>

名称解释

  • Ring Buffer
  • 环境的缓存区,3.0版本以前被认为是Disruptor的主要成员。3.0版本以后，环形缓冲区只负责通过Disruptor的事件方式来对数据进行存储和更新。在一些高级的应用场景中，Ring Buffer可以由用户的自定义实现完全替代。
  • Sequence
  • Disruptor使用Sequence作为一种方法来确定特定组件的位置。每个使用者（EventProcessor）与Disruptor本身一样维护一个序列。大多数并发代码依赖于这些序列值的移动，因此序列支持AtomicLong的许多当前特性。事实上，两者之间唯一真正的区别是序列包含额外的功能，以防止序列和其他值之间的错误共享。
  • Sequencer
  • Sequencer是真正的核心，该接口的两个实现（单生产者， 多消费者）实现了所有用于在生产者和使用者之间的快速、正确的传递数据的并发算法。
  • Sequence Barrier
  • 序列屏障由Sequencer产生，包含对Sequencer和任何依赖消费者的序列的引用。它包含确定是否有任何事件可供使用者处理的逻辑。
  • Wait Strategy
  • 等待策略确定消费者将如何等待生产者产生的消息，Disruptor将消息放到事件（Event）中。
  • Event
  • 从生产者到消费者的数据单位。不存在完全由用户定义的事件的特定代码的表示形式。
  • EventProcessor
  • EventProcessor持有特定消费者（Consumer）的Sequence，并提供用于调用事件处理实现的事件循环。
  • BatchEventProcessor
  • BatchEventProcessor它包含事件循环的有效实现，并将回调到已使用的EventHandle接口实现。
  • EventHandler
  • Disruptor定义的事件处理接口，由用户实现，用于处理事件，是Consumer的真正实现。
  • Producer
  • 生产者，只是泛指调用Disruptor发布事件的用户代码，Disruptor没有定义特定接口或类型。

架构图

简单实用Disruptor

1 定义事件

事件就是通过Disruptor进行交换的数据类型。

package com.disruptor;

public class Data {

    private long value;

    public long getValue() {
        return value;
    }

    public void setValue(long value) {
        this.value = value;
    }
}

2 定义事件工厂

事件工厂定义了如何实例化第一步中定义的事件。Disruptor通过EventFactory在RingBuffer中预创建Event的实例。

一个Event实例被用作一个数据槽，发布者发布前，先从RingBuffer获得一个Event的实例，然后往Event实例中插入数据，然后再发布到RingBuffer中，最后由Consumer获得Event实例并从中读取数据。

package com.disruptor;

import com.lmax.disruptor.EventFactory;

public class DataFactory implements EventFactory<Data> {

    @Override
    public Data newInstance() {
        return new Data();
    }
}

3 定义生产者

package com.disruptor;

import com.lmax.disruptor.RingBuffer;

import java.nio.ByteBuffer;

public class Producer {

    private final RingBuffer<Data> ringBuffer;

    public Producer(RingBuffer<Data> ringBuffer) {
        this.ringBuffer = ringBuffer;
    }

    public void pushData(ByteBuffer byteBuffer) {
        long sequence = ringBuffer.next();

        try {
            Data even = ringBuffer.get(sequence);
            even.setValue(byteBuffer.getLong(0));
        } finally {
            ringBuffer.publish(sequence);
        }
    }
}

4 定义消费者

package com.disruptor;

import com.lmax.disruptor.WorkHandler;

import java.text.MessageFormat;


public class Consumer implements WorkHandler<Data> {

    @Override
    public void onEvent(Data data) throws Exception {
        long result = data.getValue() + 1;

        System.out.println(MessageFormat.format("Data process : {0} + 1 = {1}", data.getValue(), result));
    }
}

5 启动Disruptor

  • 测试Demo
package com.disruptor;

import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;

import java.nio.ByteBuffer;
import java.util.concurrent.ThreadFactory;


public class Main {

    private static final int NUMS = 10;

    private static final int SUM = 1000000;

    public static void main(String[] args) {
        try {
            Thread.sleep(10000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        long start = System.currentTimeMillis();

        DataFactory factory = new DataFactory();

        int buffersize = 1024;

        Disruptor<Data> disruptor = new Disruptor<Data>(factory, buffersize, new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r);
            }
        });

        Consumer[] consumers = new Consumer[NUMS];
        for (int i = 0; i < NUMS; i++) {
            consumers[i] = new Consumer();
        }

        disruptor.handleEventsWithWorkerPool(consumers);
        disruptor.start();

        RingBuffer<Data> ringBuffer = disruptor.getRingBuffer();
        Producer producer = new Producer(ringBuffer);

        ByteBuffer bb = ByteBuffer.allocate(8);
        for (long i = 0; i < SUM; i++) {
            bb.putLong(0, i);
            producer.pushData(bb);
            System.out.println("Success producer data : " + i);
        }
        long end = System.currentTimeMillis();

        disruptor.shutdown();

        System.out.println("Total time : " + (end - start));
    }
}
  • 结果（部分结果展示）
Data process : 999,987 + 1 = 999,988
Success producer data : 999995
Data process : 999,990 + 1 = 999,991
Data process : 999,989 + 1 = 999,990
Data process : 999,991 + 1 = 999,992
Data process : 999,992 + 1 = 999,993
Data process : 999,993 + 1 = 999,994
Data process : 999,995 + 1 = 999,996
Success producer data : 999996
Success producer data : 999997
Success producer data : 999998
Success producer data : 999999
Data process : 999,994 + 1 = 999,995
Data process : 999,996 + 1 = 999,997
Data process : 999,997 + 1 = 999,998
Data process : 999,998 + 1 = 999,999
Data process : 999,999 + 1 = 1,000,000
Total time : 14202

由结果展示可见，边生产、边消费。

彩蛋

1 事件转换类

package com.mm.demo.disruptor.translator;

import com.lmax.disruptor.EventTranslatorOneArg;
import com.mm.demo.disruptor.entity.Data;

public class DataEventTranslator implements EventTranslatorOneArg<Data, Long> {

    @Override
    public void translateTo(Data event, long sequence, Long arg0) {
        System.out.println(MessageFormat.format("DataEventTranslator arg0 = {0}, seq = {1}", arg0, sequence));
        event.setValue(arg0);
    }
}

2 消费者

2.1 消费者Demo1

消费者每次将event的结果加1。

package com.mm.demo.disruptor.handler;

import com.lmax.disruptor.EventHandler;
import com.mm.demo.disruptor.entity.Data;

import java.text.MessageFormat;

public class D1DataEventHandler implements EventHandler<Data> {

    @Override
    public void onEvent(Data event, long sequence, boolean endOfBatch) throws Exception {
        long result = event.getValue() + 1;
        Thread t = new Thread();
        String name = t.getName();
        System.out.println(MessageFormat.format("consumer "+name+": {0} + 1 = {1}", event.getValue(), result));
    }

}

这里是使用的是EventHandler。也是使用WorkHandler，EventHandler和WorkHandler的区别是前者不需要池化，后者需要池化。

2.2 消费者Demo2

package com.mm.demo.disruptor.handler;

import com.lmax.disruptor.EventHandler;
import com.mm.demo.disruptor.entity.Data;

import java.text.MessageFormat;


public class D2DataEventHandler implements EventHandler<Data> {

    @Override
    public void onEvent(Data event, long sequence, boolean endOfBatch) throws Exception {
        long result = event.getValue() + 2;
        System.out.println(MessageFormat.format("consumer 2: {0} + 2 = {1}", event.getValue(), result));
    }
}

2.3 串行依次计算

Consumer1执行完成再执行Consumer2。

package com.mm.demo.disruptor.process;

import com.lmax.disruptor.dsl.Disruptor;
import com.mm.demo.disruptor.entity.Data;
import com.mm.demo.disruptor.handler.D1DataEventHandler;
import com.mm.demo.disruptor.handler.D2DataEventHandler;

/**
 * 串行依次计算
 * @DateT: 2020-01-07
 */
public class Serial {

    public static void serial(Disruptor<Data> disruptor) {
        disruptor.handleEventsWith(new D1DataEventHandler()).then(new D2DataEventHandler());
        disruptor.start();
    }
}

2.4 并行实时计算

Consumer1和Consumer2同时执行。

package com.mm.demo.disruptor.process;

import com.lmax.disruptor.dsl.Disruptor;
import com.mm.demo.disruptor.entity.Data;
import com.mm.demo.disruptor.handler.D1DataEventHandler;
import com.mm.demo.disruptor.handler.D2DataEventHandler;

/**
 * 并行执行
 * @DateT: 2020-01-07
 */
public class Parallel {

    public static void parallel(Disruptor<Data> dataDisruptor) {
        dataDisruptor.handleEventsWith(new D1DataEventHandler(), new D2DataEventHandler());
        dataDisruptor.start();
    }
}

2.5 测试类

package com.mm.demo.disruptor;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.mm.demo.disruptor.entity.Data;
import com.mm.demo.disruptor.handler.D1DataEventHandler;
import com.mm.demo.disruptor.process.Parallel;
import com.mm.demo.disruptor.process.Serial;
import com.mm.demo.disruptor.translator.DataEventTranslator;

import javax.swing.plaf.synth.SynthTextAreaUI;
import java.nio.ByteBuffer;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;


public class Main {

    private static final int BUFFER = 1024 * 1024;

    public static void main(String[] args) {

        DataFactory factory = new DataFactory();

        Disruptor<Data> disruptor = new Disruptor<Data>(factory, BUFFER, Executors.defaultThreadFactory(), ProducerType.MULTI, new BlockingWaitStrategy());

      
        Serial.serial(disruptor);
//        Parallel.parallel(disruptor);

        RingBuffer<Data> ringBuffer = disruptor.getRingBuffer();
        for (int i = 0; i < 2; i++) {
            ringBuffer.publishEvent(new DataEventTranslator(), (long)i);
        }
        disruptor.shutdown();
    }
}

总结

上边只演示了串行和并行的方式，其实还是通过组合的方式创建不的计算处理方式（需要创建多个事件处理器EventHandler）。

补充等待策略

  • BlockingWaitStrategy：最低效的策略，但是对cpu的消耗是最小的，在各种不同部署环境中能提供更加一致的性能表现。
  • SleepingWaitStrategy：性能和BlockingWaitStrategy差不多少，cpu消耗也类似，但是其对生产者线程的影响最小，适合用于异步处理数据的场景。
  • YieldingWaitStrategy：性能是最好的，适用于低延迟的场景。在要求极高性能且事件处理线程数小于cpu处理核数时推荐使用此策略。
  • BusySpinWaitStrategy：低延迟，但是对cpu资源的占用较多。
  • PhasedBackoffWaitStrategy：上边几种策略的综合体，延迟大，但是占用cpu资源较少。

相关推荐

为何越来越多的编程语言使用JSON（为什么编程）

JSON是JavascriptObjectNotation的缩写，意思是Javascript对象表示法，是一种易于人类阅读和对编程友好的文本数据传递方法，是JavaScript语言规范定义的一个子...

何时在数据库中使用 JSON（数据库用json格式存储）

在本文中，您将了解何时应考虑将JSON数据类型添加到表中以及何时应避免使用它们。每天?分享?最新?软件?开发?，Devops，敏捷?，测试?以及?项目?管理?最新?，最热门?的?文章?，每天?花?...

MySQL 从零开始:05 数据类型（mysql数据类型有哪些,并举例）

前面的讲解中已经接触到了表的创建，表的创建是对字段的声明，比如：上述语句声明了字段的名称、类型、所占空间、默认值和是否可以为空等信息。其中的int、varchar、char和decimal都...

JSON对象花样进阶（json格式对象）

一、引言在现代Web开发中，JSON（JavaScriptObjectNotation）已经成为数据交换的标准格式。无论是从前端向后端发送数据，还是从后端接收数据，JSON都是不可或缺的一部分。...

深入理解 JSON 和 Form-data（json和formdata提交区别）

在讨论现代网络开发与API设计的语境下，理解客户端和服务器间如何有效且可靠地交换数据变得尤为关键。这里，特别值得关注的是两种主流数据格式：...

JSON 语法（json 语法 priority）

JSON语法是JavaScript语法的子集。JSON语法规则JSON语法是JavaScript对象表示法语法的子集。数据在名称/值对中数据由逗号分隔花括号保存对象方括号保存数组JS...

JSON语法详解（json的语法规则）

JSON语法规则JSON语法是JavaScript对象表示法语法的子集。数据在名称/值对中数据由逗号分隔大括号保存对象中括号保存数组注意：json的key是字符串,且必须是双引号，不能是单引号...

MySQL JSON数据类型操作（mysql的json）

概述mysql自5.7.8版本开始，就支持了json结构的数据存储和查询，这表明了mysql也在不断的学习和增加nosql数据库的有点。但mysql毕竟是关系型数据库，在处理json这种非结构化的数据...

JSON的数据模式（json数据格式示例）

像XML模式一样，JSON数据格式也有Schema，这是一个基于JSON格式的规范。JSON模式也以JSON格式编写。它用于验证JSON数据。JSON模式示例以下代码显示了基本的JSON模式。{"...

前端学习——JSON格式详解（后端json格式）

JSON(JavaScriptObjectNotation)是一种轻量级的数据交换格式。易于人阅读和编写。同时也易于机器解析和生成。它基于JavaScriptProgrammingLa...

什么是 JSON:详解 JSON 及其优势（什么叫json）

现在程序员还有谁不知道JSON吗？无论对于前端还是后端，JSON都是一种常见的数据格式。那么JSON到底是什么呢？JSON的定义...

PostgreSQL JSON 类型:处理结构化数据

PostgreSQL提供JSON类型，以存储结构化数据。JSON是一种开放的数据格式，可用于存储各种类型的值。什么是JSON类型？JSON类型表示JSON(JavaScriptO...

JavaScript:JSON、三种包装类（javascript 包）

JOSN:我们希望可以将一个对象在不同的语言中进行传递，以达到通信的目的，最佳方式就是将一个对象转换为字符串的形式JSON（JavaScriptObjectNotation）-JS的对象表示法...

Python数据分析 只要1分钟 教你玩转JSON 全程干货

Json简介：Json，全名JavaScriptObjectNotation,JSON(JavaScriptObjectNotation(记号、标记))是一种轻量级的数据交换格式。它基于J...

比较一下JSON与XML两种数据格式?（json和xml哪个好）

JSON（JavaScriptObjectNotation）和XML（eXtensibleMarkupLanguage）是在日常开发中比较常用的两种数据格式，它们主要的作用就是用来进行数据的传...

取消回复欢迎 你 发表评论:

请填写验证码
一周热门
最近发表
标签列表