介绍

在使用中会遇到，多个生产者将事件推给消费者的情况

实现代码

• 创建一个事件类
  用于生产者和消费者之间进行通讯的的事件

public class TradeEvent {
    private Integer userId;
    private String id;
    private String name;
    private Integer status;
    private Integer produceId;

    public Integer getProduceId() {
        return produceId;
    }

    public void setProduceId(Integer produceId) {
        this.produceId = produceId;
    }

    public Integer getUserId() {
        return userId;
    }

    public void setUserId(Integer userId) {
        this.userId = userId;
    }

    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public Integer getStatus() {
        return status;
    }

    public void setStatus(Integer status) {
        this.status = status;
    }

    @Override
    public String toString() {
        return "TradeEvent{" +
                "userId=" + userId +
                ", id='" + id + '\'' +
                ", name='" + name + '\'' +
                ", status=" + status +
                ", produceId=" + produceId +
                '}';
    }
}

• 生产者类
  用来生产事件的，可以创建多个实例来进行模拟多生产者的场景

import com.lmax.disruptor.RingBuffer;

import java.nio.ByteBuffer;

public class Producer {
    public static void translate(TradeEvent event, long sequence, ByteBuffer buffer) {
        event.setUserId(buffer.getInt(0));
        event.setProduceId(buffer.getInt(4));
    }
    private RingBuffer<TradeEvent> ringBuffer;
    private Integer produceId;

    public Producer(RingBuffer<TradeEvent> ringBuffer, Integer produceId) {
        this.ringBuffer = ringBuffer;
        this.produceId = produceId;
    }

    public void process(){
        ByteBuffer bb = ByteBuffer.allocate(8);
        for (int l = 1; true; l++) {
            bb.putInt(0, l);
            bb.putInt(4,produceId);
            ringBuffer.publishEvent(Producer::translate, bb);
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

• 启动类

import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.lmax.disruptor.util.DaemonThreadFactory;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class LongEventMain4 {
    public static void handleEvent(TradeEvent event, long sequence, boolean endOfBatch) {
        System.out.println(event);
    }


    public static void main(String[] args) throws Exception {
        // Specify the size of the ring buffer, must be power of 2.
        int bufferSize = 1024;

        // Construct the Disruptor
        Disruptor<TradeEvent> disruptor = new Disruptor<>(TradeEvent::new,
                bufferSize,
                DaemonThreadFactory.INSTANCE,
                ProducerType.MULTI,
                new YieldingWaitStrategy());

        // Connect the handler
        disruptor.handleEventsWith(LongEventMain::handleEvent);

        // Start the Disruptor, starts all threads running
        disruptor.start();

        // Get the ring buffer from the Disruptor to be used for publishing.
        RingBuffer<TradeEvent> ringBuffer = disruptor.getRingBuffer();

        //模拟多个生产者
        ExecutorService threadPool = Executors.newFixedThreadPool(3);
        for (int i = 0; i < 3; i++) {
            final int a = i;
            threadPool.execute(() ->
                    //将ringBuffer 和 生产者ID传入
                    new Producer(ringBuffer, a).process()
            );
        }
    }
}