基于disruptor实现简单topic分发消息功能
前言
disruptor 性能的确很强 但是只是做了 队列的功能 如果有多种消息 就必须自己去扩展一下 或者用多个队列
自己手写了一套简易的 基于disruptor 点对点的 topic分发功能
思路
通过一个分发处理器 将收到消息按照topic 分发到不同的处理器 所有的消息由DistributeEventHandler 按照DisruptorTopicEnum 进行分发到不同的BaseEventHandler实现类
发送消息通过DisruptorService
示例
包结构如下图:
- Element
package com.ming.core.disruptor;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
/**
* 测试disruptor的element
*
* @author ming
* @date 2020-10-27 16:17
*/
@Slf4j
@Data
public class Element<T> {
private DisruptorTopicEnum topic;
private T data;
}
- DisruptorService
package com.ming.core.disruptor;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Primary;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.concurrent.ThreadFactory;
/**
* disruptor 服务
*
* @author ming
* @date 2020-10-27 17:03
*/
@Primary
@Component
@Slf4j
public class DisruptorService {
@Autowired
private DistributeEventHandler distributeEventHandler;
private Disruptor<Element> elementDisruptor;
/**
* 发送消息
*
* @author ming
* @date 2020-10-27 17:08
*/
@SuppressWarnings("unchecked")
public <T> void sendMessage(DisruptorTopicEnum disruptorTopicEnum, T data) {
RingBuffer<Element> ringBuffer = elementDisruptor.getRingBuffer();
// 获取下一个可用位置的下标
long sequence = ringBuffer.next();
try {
// 返回可用位置的元素
Element<T> event = ringBuffer.get(sequence);
// 设置该位置元素的值
event.setData(data);
event.setTopic(disruptorTopicEnum);
} finally {
ringBuffer.publish(sequence);
}
}
/**
* 初始化 disruptor队列
*
* @author ming
* @date 2020-10-27 17:12
*/
@PostConstruct
public void init() {
//初始化disruptor
// 生产者的线程工厂
ThreadFactory threadFactory = r -> new Thread(r, "simpleThread");
// RingBuffer生产工厂,初始化RingBuffer的时候使用
EventFactory<Element> factory = Element::new;
// 阻塞策略
BlockingWaitStrategy strategy = new BlockingWaitStrategy();
// 指定RingBuffer的大小
int bufferSize = 16;
// 创建disruptor，采用单生产者模式
elementDisruptor = new Disruptor(factory, bufferSize, threadFactory, ProducerType.SINGLE, strategy);
// 设置EventHandler 并且后置清理消费过的数据
elementDisruptor.handleEventsWith(distributeEventHandler);
elementDisruptor.handleExceptionsFor(distributeEventHandler).with(disruptorExceptionHandler);
// 启动disruptor的线程
elementDisruptor.start();
}
/**
* 销毁 disruptor队列
*
* @author ming
* @date 2020-10-27 17:12
*/
@PreDestroy
public void destroy() {
//销毁 disruptor
elementDisruptor.shutdown();
}
}
- DisruptorTopicEnum
package com.ming.core.disruptor;
/**
* disruptor topic 枚举
*
* @author ming
* @date 2020-10-27 16:58
*/
public enum DisruptorTopicEnum {
LOG_TOPIC, TEST1_TOPIC, TEST2_TOPIC;
}
- DistributeEventHandler
package com.ming.core.disruptor;
import com.lmax.disruptor.EventHandler;
import com.ming.core.disruptor.handler.LogEventHandler;
import com.ming.core.disruptor.handler.Test1EventHandler;
import com.ming.core.disruptor.handler.Test2EventHandler;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
/**
* 基础event handler
* 根据topic 分发数据
*
* @author ming
* @date 2020-10-27 17:15
*/
@Component
public class DistributeEventHandler implements EventHandler {
@Autowired
private Test1EventHandler test1EventHandler;
@Autowired
private Test2EventHandler test2EventHandler;
@Autowired
private LogEventHandler logEventHandler;
/**
* Called when a publisher has published an event to the {@link RingBuffer}. The {@link BatchEventProcessor} will
* read messages from the {@link RingBuffer} in batches, where a batch is all of the events available to be
* processed without having to wait for any new event to arrive. This can be useful for event handlers that need
* to do slower operations like I/O as they can group together the data from multiple events into a single
* operation. Implementations should ensure that the operation is always performed when endOfBatch is true as
* the time between that message an the next one is inderminate.
*
* @param event published to the {@link RingBuffer}
* @param sequence of the event being processed
* @param endOfBatch flag to indicate if this is the last event in a batch from the {@link RingBuffer}
* @throws Exception if the EventHandler would like the exception handled further up the chain.
*/
@Override
@SuppressWarnings("unchecked")
public void onEvent(Object event, long sequence, boolean endOfBatch) throws Exception {
if (!(event instanceof Element)) {
throw new RuntimeException("类型错误,必须为Element类型~");
}
Element<?> element = (Element<?>) event;
//分发topic 到对应的handler
switch (element.getTopic()) {
case LOG_TOPIC -> logEventHandler.onEvent((Element<String>) event, sequence, endOfBatch);
case TEST1_TOPIC -> test1EventHandler.onEvent((Element<String>) event, sequence, endOfBatch);
case TEST2_TOPIC -> test2EventHandler.onEvent((Element<String>) event, sequence, endOfBatch);
default -> throw new RuntimeException("topic未注册!无法分发消息");
}
}
}
- DisruptorExceptionHandler
package com.ming.core.disruptor;
import com.lmax.disruptor.ExceptionHandler;
import com.ming.core.utils.JacksonJsonSingleton;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
/**
* disruptor 异常处理
*
* @author ming
* @date 2020-10-28 11:32
*/
@Component
@Slf4j
public class DisruptorExceptionHandler<T> implements ExceptionHandler<T> {
/**
* <p>Strategy for handling uncaught exceptions when processing an event.</p>
*
* <p>If the strategy wishes to terminate further processing by the {@link BatchEventProcessor}
* then it should throw a {@link RuntimeException}.</p>
*
* @param ex the exception that propagated from the {@link EventHandler}.
* @param sequence of the event which cause the exception.
* @param event being processed when the exception occurred. This can be null.
*/
@Override
public void handleEventException(Throwable ex, long sequence, Object event) {
log.error("处理disruptor事件异常,异常内容:{},编号:{},事件内容:{}", ex.getMessage(), sequence, JacksonJsonSingleton.writeString(event));
ex.printStackTrace();
}
/**
* Callback to notify of an exception during {@link LifecycleAware#onStart()}
*
* @param ex throw during the starting process.
*/
@Override
public void handleOnStartException(Throwable ex) {
log.error("处理disruptor启动异常,异常内容:{}", ex.getMessage());
ex.printStackTrace();
}
/**
* Callback to notify of an exception during {@link LifecycleAware#onShutdown()}
*
* @param ex throw during the shutdown process.
*/
@Override
public void handleOnShutdownException(Throwable ex) {
log.error("处理disruptor关闭异常,异常内容:{}", ex.getMessage());
ex.printStackTrace();
}
}
- BaseEventHandler<T, R>
package com.ming.core.disruptor.handler;
import com.ming.core.disruptor.Element;
/**
* 基础 事件处理接口
*
* @author ming
* @date 2020-10-28 09:37
*/
public interface BaseEventHandler<T, R> {
R onEvent(Element<T> event, long sequence, boolean endOfBatch);
}
- LogEventHandler
package com.ming.core.disruptor.handler;
import com.ming.core.disruptor.Element;
import org.springframework.stereotype.Component;
/**
* 基础event handler
* 根据topic 分发数据
*
* @author ming
* @date 2020-10-27 17:15
*/
@Component
public class LogEventHandler implements BaseEventHandler<String, String> {
@Override
public String onEvent(Element<String> event, long sequence, boolean endOfBatch) {
System.out.println("log:" + event.getData());
return "log";
}
}
- Test1EventHandler
package com.ming.core.disruptor.handler;
import com.ming.core.disruptor.Element;
import org.springframework.stereotype.Component;
/**
* 基础event handler
* 根据topic 分发数据
*
* @author ming
* @date 2020-10-27 17:15
*/
@Component
public class Test1EventHandler implements BaseEventHandler<String, String> {
@Override
public String onEvent(Element<String> event, long sequence, boolean endOfBatch) {
System.out.println("Test1:" + event.getData());
return "Test1";
}
}
- Test2EventHandler
package com.ming.core.disruptor.handler;
import com.ming.core.disruptor.Element;
import org.springframework.stereotype.Component;
/**
* 基础event handler
* 根据topic 分发数据
*
* @author ming
* @date 2020-10-27 17:15
*/
@Component
public class Test2EventHandler implements BaseEventHandler<String, String> {
@Override
public String onEvent(Element<String> event, long sequence, boolean endOfBatch) {
System.out.println("Test2:" + event.getData());
return "Test2";
}
}
- 测试用例
package com.ming.core.disruptor;
import com.ming.Start;
import org.junit.jupiter.api.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
/**
* disruptor service test
*
* @author ming
* @date 2020-10-28 10:21
*/
@RunWith(SpringRunner.class)
@SpringBootTest(classes = Start.class)
class DisruptorServiceTest {
@Autowired
private DisruptorService disruptorService;
@Test
void sendMessage() {
int max = 10000;
long now = System.currentTimeMillis();
new Thread(() -> {
for (int i = 0; i < max; i++) {
disruptorService.sendMessage(DisruptorTopicEnum.LOG_TOPIC, i + "minglog" + (System.currentTimeMillis()-now));
}
}).start();
new Thread(() -> {
for (int i = 0; i < max; i++) {
disruptorService.sendMessage(DisruptorTopicEnum.TEST1_TOPIC, i + "mingtest1" + (System.currentTimeMillis()-now));
}
}).start();
new Thread(() -> {
for (int i = 0; i < max; i++) {
disruptorService.sendMessage(DisruptorTopicEnum.TEST2_TOPIC, i + "mingtest2" + (System.currentTimeMillis()-now));
}
}).start();
try {
Thread.sleep(10000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
总结
disruptor 性能是很高 但是他本身只是负责队列部分 大多数场景需要根据具体需求进行扩展调整 大多数用jdk的queue的 都可以用disruptor替换