使用Disruptor完成多个消费者不重复消费消息
使用Disruptor完成多个消费者不重复消费消息
天涯泪小武
发布于 2019-01-17 12:01:10
发布于 2019-01-17 12:01:10
上一篇https://blog.csdn.net/tianyaleixiaowu/article/details/79787377里讲了Disruptor完成多个消费者并行、顺序重复消费Event。重复消费类似于kafka中，同一个topic被不同的group的消费者消费。这样的场景比较常见。当然更常见的场景是不重复消费，也就是一个消息只能被消费一次。
Disruptor同样可以完成不重复消费的功能。
上一篇消费者消费Handler是这样的，需要实现EventHandler。
/**
* @author wuweifeng wrote on 2018/3/29.
*/
public class LastEventHandler implements EventHandler<LongEvent> {
@Override
public void onEvent(LongEvent secondEvent, long sequence, boolean endOfBatch) throws Exception {
System.out.println("-------Last:" + secondEvent);
}
}这次的消费者需要实现不同的接口，代码如下，注意，代码是接续上一篇的，需要先下载上一篇的代码。
import a.LongEvent;
import com.lmax.disruptor.WorkHandler;
/**
* @author wuweifeng wrote on 2018/4/8.
*/
public class Consumer implements WorkHandler<LongEvent> {
@Override
public void onEvent(LongEvent longEvent) throws Exception {
System.out.println(Thread.currentThread().getName() + "消费者消费了消息：" + longEvent.toString());
}
}import a.LongEvent;
import a.LongEventFactory;
import a.LongEventProducer;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import java.nio.ByteBuffer;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
/**
* @author wuweifeng wrote on 2018/4/8.
*/
@SuppressWarnings("ALL")
public class Test {
public static void main(String[] args) {
ThreadFactory producerFactory = Executors.defaultThreadFactory();
// 创建缓冲池
LongEventFactory eventFactory = new LongEventFactory();
// 创建bufferSize ,也就是RingBuffer大小，必须是2的N次方
int ringBufferSize = 1024 * 1024;
Disruptor<LongEvent> disruptor = new Disruptor<>(eventFactory, ringBufferSize, producerFactory,
ProducerType.SINGLE, new BlockingWaitStrategy());
RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
// 创建10个消费者来处理同一个生产者发的消息(这10个消费者不重复消费消息)
Consumer[] consumers = new Consumer[10];
for (int i = 0; i < consumers.length; i++) {
consumers[i] = new Consumer();
}
disruptor.handleEventsWithWorkerPool(consumers);
disruptor.start();
LongEventProducer longEventProducer = new LongEventProducer(ringBuffer);
ByteBuffer bb = ByteBuffer.allocate(8);
for (long i = 0; i < 100L; i++) {
bb.putLong(0, i);
longEventProducer.onData(bb);
}
disruptor.shutdown();
}
}这个是测试类，可以发现和上一篇的主要区别就是disruptor.handleEventsWithWorkerPool。其他的都是一样的。
使用handleEventsWithWorkerPool就可以完成不重复消费，使用handleEventsWith就是重复消费。这里定义了10个消费者，那么就会启动10个线程来不重复消费生产者发出的100条消息。
本文参与 腾讯云自媒体分享计划，分享自作者个人站点/博客。
原始发表：2018年04月08日，如有侵权请联系 cloudcommunity@tencent.com 删除
评论
登录后参与评论
推荐阅读