disruptor笔记之三：环形队列的基础操作(不用Disruptor类)

Posted 2021-06-14 程序员欣宸

欢迎访问我的GitHub

这里分类和汇总了欣宸的全部原创(含配套源码)：https://github.com/zq2599/blog_demos

本篇概览

• 本文是《disruptor笔记》系列的第三篇，主要任务是编码实现消息生产和消费，与《disruptor笔记之一：快速入门》不同的是，本次开发不使用Disruptor类，和Ring Buffer(环形队列)相关的操作都是自己写代码实现；
• 这种脱离Disruptor类操作Ring Buffer的做法，不适合用在生产环境，但在学习Disruptor的过程中，这是种高效的学习手段，经过本篇实战后，在今后使用Disruptor时，您在开发、调试、优化等各种场景下都能更加得心应手；
• 简单的消息生产消费已不能满足咱们的学习热情，今天的实战要挑战以下三个场景：

1. 100个事件，单个消费者消费；
2. 100个事件，三个消费者，每个都独自消费这个100个事件；
3. 100个事件，三个消费者共同消费这个100个事件；

前文回顾

为了完成本篇的实战，前文《disruptor笔记之二：Disruptor类分析》已做了充分的研究分析，建议观看，这里简单回顾以下Disruptor类的几个核心功能，这也是咱们编码时要实现的：

1. 创建环形队列（RingBuffer对象）
2. 创建SequenceBarrier对象，用于接收ringBuffer中的可消费事件
3. 创建BatchEventProcessor，负责消费事件
4. 绑定BatchEventProcessor对象的异常处理类
5. 调用ringBuffer.addGatingSequences，将消费者的Sequence传给ringBuffer
6. 启动独立线程，用来执行消费事件的业务逻辑

• 理论分析已经完成，接下来开始编码；

源码下载

• 本篇实战中的完整源码可在GitHub下载到，地址和链接信息如下表所示(https://github.com/zq2599/blog_demos)：

名称	链接	备注
项目主页	https://github.com/zq2599/blog_demos	该项目在GitHub上的主页
git仓库地址(https)	https://github.com/zq2599/blog_demos.git	该项目源码的仓库地址，https协议
git仓库地址(ssh)	[e-mail]:zq2599/blog_demos.git	该项目源码的仓库地址，ssh协议

• 这个git项目中有多个文件夹，本次实战的源码在disruptor-tutorials文件夹下，如下图红框所示：
  
• disruptor-tutorials是个父工程，里面有多个module，本篇实战的module是low-level-operate，如下图红框所示：
  

开发

• 进入编码阶段，今天的任务是挑战以下三个场景：

1. 100个事件，单个消费者消费；
2. 100个事件，三个消费者，每个都独自消费这个100个事件；
3. 100个事件，三个消费者共同消费这个100个事件；

• 咱们先把工程建好，然后编写公共代码，例如事件定义、事件工厂等，最后才是每个场景的开发；
• 在父工程disruptor-tutorials新增名为low-level-operate的module，其build.gradle如下：

plugins {
    id 'org.springframework.boot'
}

dependencies {
    implementation 'org.projectlombok:lombok'
    implementation 'org.springframework.boot:spring-boot-starter'
    implementation 'org.springframework.boot:spring-boot-starter-web'
    implementation 'com.lmax:disruptor'

    testImplementation('org.springframework.boot:spring-boot-starter-test')
}

• 然后是springboot启动类：

package com.bolingcavalry;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class LowLevelOperateApplication {
	public static void main(String[] args) {
		SpringApplication.run(LowLevelOperateApplication.class, args);
	}
}

• 事件类，这是事件的定义：

package com.bolingcavalry.service;

import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.ToString;

@Data
@ToString
@NoArgsConstructor
public class StringEvent {
    private String value;
}

• 事件工厂，定义如何在内存中创建事件对象：

package com.bolingcavalry.service;

import com.lmax.disruptor.EventFactory;

public class StringEventFactory implements EventFactory<StringEvent> {
    @Override
    public StringEvent newInstance() {
        return new StringEvent();
    }
}

• 事件生产类，定义如何将业务逻辑的事件转为disruptor事件发布到环形队列，用于消费：

package com.bolingcavalry.service;

import com.lmax.disruptor.RingBuffer;

public class StringEventProducer {

    // 存储数据的环形队列
    private final RingBuffer<StringEvent> ringBuffer;

    public StringEventProducer(RingBuffer<StringEvent> ringBuffer) {
        this.ringBuffer = ringBuffer;
    }

    public void onData(String content) {

        // ringBuffer是个队列，其next方法返回的是下最后一条记录之后的位置，这是个可用位置
        long sequence = ringBuffer.next();

        try {
            // sequence位置取出的事件是空事件
            StringEvent stringEvent = ringBuffer.get(sequence);
            // 空事件添加业务信息
            stringEvent.setValue(content);
        } finally {
            // 发布
            ringBuffer.publish(sequence);
        }
    }
}

• 事件处理类，收到事件后具体的业务处理逻辑：

package com.bolingcavalry.service;

import com.lmax.disruptor.EventHandler;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import java.util.function.Consumer;

@Slf4j
public class StringEventHandler implements EventHandler<StringEvent> {

    public StringEventHandler(Consumer<?> consumer) {
        this.consumer = consumer;
    }

    // 外部可以传入Consumer实现类，每处理一条消息的时候，consumer的accept方法就会被执行一次
    private Consumer<?> consumer;

    @Override
    public void onEvent(StringEvent event, long sequence, boolean endOfBatch) throws Exception {
        log.info("sequence [{}], endOfBatch [{}], event : {}", sequence, endOfBatch, event);

        // 这里延时100ms，模拟消费事件的逻辑的耗时
        Thread.sleep(100);

        // 如果外部传入了consumer，就要执行一次accept方法
        if (null!=consumer) {
            consumer.accept(null);
        }
    }
}

• 定义一个接口，外部通过调用接口的方法来生产消息，再放几个常量在里面后面会用到：

package com.bolingcavalry.service;

public interface LowLevelOperateService {
    /**
     * 消费者数量
     */
    int CONSUMER_NUM = 3;

    /**
     * 环形缓冲区大小
     */
    int BUFFER_SIZE = 16;

    /**
     * 发布一个事件
     * @param value
     * @return
     */
    void publish(String value);

    /**
     * 返回已经处理的任务总数
     * @return
     */
    long eventCount();
}

• 以上就是公共代码了，接下来逐个实现之前规划的三个场景；

100个事件，单个消费者消费

• 这是最简单的功能了，实现发布消息和单个消费者消费的功能，代码如下，有几处要注意的地方稍后提到：

package com.bolingcavalry.service.impl;

import com.bolingcavalry.service.*;
import com.lmax.disruptor.BatchEventProcessor;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SequenceBarrier;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;

@Service("oneConsumer")
@Slf4j
public class OneConsumerServiceImpl implements LowLevelOperateService {

    private RingBuffer<StringEvent> ringBuffer;

    private StringEventProducer producer;

    /**
     * 统计消息总数
     */
    private final AtomicLong eventCount = new AtomicLong();

    private ExecutorService executors;

    @PostConstruct
    private void init() {
        // 准备一个匿名类，传给disruptor的事件处理类，
        // 这样每次处理事件时，都会将已经处理事件的总数打印出来
        Consumer<?> eventCountPrinter = new Consumer<Object>() {
            @Override
            public void accept(Object o) {
                long count = eventCount.incrementAndGet();
                log.info("receive [{}] event", count);
            }
        };

        // 创建环形队列实例
        ringBuffer = RingBuffer.createSingleProducer(new StringEventFactory(), BUFFER_SIZE);

        // 准备线程池
        executors = Executors.newFixedThreadPool(1);

        //创建SequenceBarrier
        SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();

        // 创建事件处理的工作类，里面执行StringEventHandler处理事件
        BatchEventProcessor<StringEvent> batchEventProcessor = new BatchEventProcessor<>(
                ringBuffer,
                sequenceBarrier,
                new StringEventHandler(eventCountPrinter));

        // 将消费者的sequence传给环形队列
        ringBuffer.addGatingSequences(batchEventProcessor.getSequence());

        // 在一个独立线程中取事件并消费
        executors.submit(batchEventProcessor);

        // 生产者
        producer = new StringEventProducer(ringBuffer);
    }

    @Override
    public void publish(String value) {
        producer.onData(value);
    }

    @Override
    public long eventCount() {
        return eventCount.get();
    }
}

• 上述代码有以下几处需要注意：

1. 自己创建环形队列RingBuffer实例
2. 自己准备线程池，里面的线程用来获取和消费消息
3. 自己动手创建BatchEventProcessor实例，并把事件处理类传入
4. 通过ringBuffer创建sequenceBarrier，传给BatchEventProcessor实例使用
5. 将BatchEventProcessor的sequence传给ringBuffer，确保ringBuffer的生产和消费不会出现混乱
6. 启动线程池，意味着BatchEventProcessor实例在一个独立线程中不断的从ringBuffer中获取事件并消费；

• 为了验证上述代码能否正常工作，我这里写了个单元测试类，如下所示，逻辑很简单，调用OneConsumerServiceImpl.publish方法一百次，产生一百个事件，再检查OneConsumerServiceImpl记录的消费事件总数是不是等于一百：

package com.bolingcavalry.service.impl;

import com.bolingcavalry.service.LowLevelOperateService;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
import static org.junit.Assert.assertEquals;

@RunWith(SpringRunner.class)
@SpringBootTest
@Slf4j
public class LowLeverOperateServiceImplTest {

    @Autowired
    @Qualifier("oneConsumer")
    LowLevelOperateService oneConsumer;

    private static final int EVENT_COUNT = 100;

    private void testLowLevelOperateService(LowLevelOperateService service, int eventCount, int expectEventCount) throws InterruptedException {
        for(int i=0;i<eventCount;i++) {
            log.info("publich {}", i);
            service.publish(String.valueOf(i));
        }

        // 异步消费，因此需要延时等待
        Thread.sleep(10000);

        // 消费的事件总数应该等于发布的事件数
        assertEquals(expectEventCount, service.eventCount());
    }

    @Test
    public void testOneConsumer() throws InterruptedException {
        log.info("start testOneConsumerService");
        testLowLevelOperateService(oneConsumer, EVENT_COUNT, EVENT_COUNT);
    }

• 注意，如果您是直接在IDEA上点击图标来执行单元测试，记得勾选下图红框中选项，否则可能出现编译失败：
  
• 执行上述单元测试类，结果如下图所示，消息的生产和消费都符合预期，并且消费逻辑是在独立线程中执行的：
  
• 继续挑战下一个场景；

100个事件，三个消费者，每个都独自消费这个100个事件

• 这个场景在kafka中也有，就是三个消费者的group不同，这样每一条消息，这两个消费者各自消费一次；
• 因此，100个事件，3个消费者每人都会独立消费这100个事件，一共消费300次；
• 代码如下，有几处要注意的地方稍后提到：

package com.bolingcavalry.service.impl;

import com.bolingcavalry.service.*;
import com.lmax.disruptor.BatchEventProcessor;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SequenceBarrier;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;

@Service("multiConsumer")
@Slf4j
public class MultiConsumerServiceImpl implements LowLevelOperateService {

    private RingBuffer<StringEvent> ringBuffer;

    private StringEventProducer producer;

    /**
     * 统计消息总数
     */
    private final AtomicLong eventCount = new AtomicLong();

    /**
     * 生产一个BatchEventProcessor实例，并且启动独立线程开始获取和消费消息
     * @param executorService
     */
    private void addProcessor(ExecutorService executorService) {
        // 准备一个匿名类，传给disruptor的事件处理类，
        // 这样每次处理事件时，都会将已经处理事件的总数打印出来
        Consumer<?> eventCountPrinter = new Consumer<Object>() {
            @Override
            public void accept(Object o) {
                long count = eventCount.incrementAndGet();
                log.info("receive [{}] event", count);
            }
        };

        BatchEventProcessor<StringEvent> batchEventProcessor = new BatchEventProcessor<>(
                ringBuffer,
                ringBuffer.newBarrier(),
                new StringEventHandler(eventCountPrinter));

        // 将当前消费者的sequence实例传给ringBuffer
        ringBuffer.addGatingSequences(batchEventProcessor.getSequence());

        // 启动独立线程获取和消费事件
        executorService.submit(batchEventProcessor);
    }

    @PostConstruct
    private void init() {
        ringBuffer = RingBuffer.createSingleProducer(new StringEventFactory(), BUFFER_SIZE);

        ExecutorService executorService = Executors.newFixedThreadPool(CONSUMER_NUM);

        // 创建多个消费者，并在独立线程中获取和消费事件
        for (int i=0;i<CONSUMER_NUM;i++) {
            addProcessor(executorService);
        }

        // 生产者
        producer = new StringEventProducer(ringBuffer);
    }

    @Override
    public void publish(String value) {
        producer.onData(value);
    }

    @Override
    public long eventCount() {
        return eventCount.get();
    }
}

• 上述代码和前面的OneConsumerServiceImpl相比差别不大，主要是创建了多个BatchEventProcessor实例，然后分别在线程池中提交；
• 验证方法依旧是单元测试，在刚才的LowLeverOperateServiceImplTest.java中增加代码即可，注意testLowLevelOperateService的第三个参数是EVENT_COUNT * LowLevelOperateService.CONSUMER_NUM，表示预期的被消费消息数为300：

 	@Autowired
    @Qualifier("multiConsumer")
    LowLevelOperateService multiConsumer;

    @Test
    public void testMultiConsumer() throws InterruptedException {
        log.info("start testMultiConsumer");
        testLowLevelOperateService(multiConsumer, EVENT_COUNT, EVENT_COUNT * LowLevelOperateService.CONSUMER_NUM);
    }

• 执行单元测试，如下图所示，一共消费了300个事件，并且三个消费者在不动线程：
  

100个事件，三个消费者共同消费这个100个事件

• 本篇的最后一个实战是发布100个事件，然后让三个消费者共同消费100个（例如A消费33个，B消费33个，C消费34个）；
• 前面用到的BatchEventProcessor是用来独立消费的，不适合多个消费者共同消费，这种多个消费共同消费的场景需要借助WorkerPool来完成，这个名字还是很形象的：一个池子里面有很多个工作者，把任务放入这个池子，工作者们每人处理一部分，大家合力将任务完成；
• 传入WorkerPool的消费者需要实现WorkHandler接口，于是新增一个实现类：

package com.bolingcavalry.service;

import com.lmax.disruptor.WorkHandler;
import lombok.extern.slf4j以上是关于disruptor笔记之三：环形队列的基础操作(不用Disruptor类)的主要内容，如果未能解决你的问题，请参考以下文章
 
 disruptor笔记之八：知识点补充(终篇)
 disruptor笔记之二：Disruptor类分析
 提升--20---Disruptor-----号称最快的消息队列
 高性能队列Disruptor使用入门，原理和代码实现
 Disruptor快速入门
 Disruptor使用