高性能队列 Disruptor 在 IM 系统中的实战
前三期我们介绍了Disruptor的典型使用场景和相关高性能原理，本期我介绍一下Disruptor在IM系统用的应用实战，IM系统即社交聊天系统，对实时性的要求非常高，非常符合Disruptor的使用场景。
本篇文章将结合实际代码，介绍如何在 IM 系统中使用 Disruptor 进行高效的消息转发。
1. Disruptor 在 IM 系统中的作用
在 IM 系统中，用户 A 发送消息给 B、C、D 时，需要根据 B、C、D 所在的服务器节点进行分组，并将消息转发到对应的节点上。为了确保高吞吐量和低延迟，我们使用 Disruptor 作为高性能队列。
2. 代码实现
2.1 初始化 Disruptor
当某个节点 nodeId 还没有对应的 RingBuffer 时，我们需要创建一个新的 Disruptor，并将其存入 ringBufferMap 中。
private final Map<String, RingBuffer<ClusterPublishEvent>> ringBufferMap = new ConcurrentHashMap<>();
public ClusterQueueService(Server server) {
this.mServer = server;
}
public void publishMessage(String nodeId, String fromUser, String clientId, String topic, byte[] payload) {
if (!ringBufferMap.containsKey(nodeId)) {
long st = System.currentTimeMillis();
synchronized (ringBufferMap){
if(!ringBufferMap.containsKey(nodeId)) {
BlockingWaitStrategy strategy = new BlockingWaitStrategy();
Disruptor<ClusterPublishEvent> disruptor = new Disruptor<>(
new ClusterPublishEventFactory(), 1024 * 1024, DaemonThreadFactory.INSTANCE,
ProducerType.SINGLE, strategy);
disruptor.handleEventsWith(new ClusterPublishEventHandler(mServer, nodeId));
disruptor.setDefaultExceptionHandler(new IgnoreExceptionHandler());
disruptor.start();
ringBufferMap.put(nodeId, disruptor.getRingBuffer());
}
}
log.info("publishMessage create RingBuffer cost:{}ms, ringBufferMap:{},size:{}", System.currentTimeMillis() - st, ringBufferMap, ringBufferMap.size());
}
RingBuffer<ClusterPublishEvent> ringBuffer = ringBufferMap.get(nodeId);
long sequence = ringBuffer.next();
// 当环形缓冲区未用完时， 返回的是空对象，否则，返回的是缓存的数据。
ClusterPublishEvent clusterEvent = ringBuffer.get(sequence);
clusterEvent.setFromUser(fromUser);
clusterEvent.setClientId(clientId);
// 此topic,是节点转发的topic: NM2R, NTF,DESTROYUSER, 只有这三种
clusterEvent.setTopic(topic);
clusterEvent.setPayload(payload);
clusterEvent.setTraceId(MDC.get(ImSvcConstants.TRACE_ID));
// 发布事件， 会触发ClusterPublishEventHandler.onEvent方法
ringBuffer.publish(sequence);
}
关键点解析：
- 采用
BlockingWaitStrategy作为等待策略，确保高效的 CPU 资源利用。 - 采用
DaemonThreadFactory.INSTANCE创建线程池，避免应用程序退出时线程未正常回收。 -
handleEventsWith设定事件处理器ClusterPublishEventHandler，用于消息处理。 -
setDefaultExceptionHandler避免异常影响消息处理流程。
2.2 按照节点转发消息
根据用户所在的服务节点，进行消息转发（发送消息事件到Disruptor）
public void publish2Receivers(Long messageId, Set<String> receivers, String exceptClientId, int pullType, String topic) {
//未绑定broker的用户默认由本中心处理
Map<String, String> allReceiverMap = new HashMap<>();
for (String receiver : receivers) {
allReceiverMap.put(receiver, localNodeId);
}
//从分布式缓存获取获取用户路由
Map<String, String> receiverMap = userRouteStore.getAll(receivers);
allReceiverMap.putAll(receiverMap);
Map<String, Set<String>> nodeMap = new HashMap<>();
//使用nodeId分组
allReceiverMap.forEach((receiver, nodeId) -> {
if (!nodeMap.containsKey(nodeId)) {
nodeMap.put(nodeId, new HashSet<>());
}
nodeMap.get(nodeId).add(receiver);
});
//获取可用节点
Cluster cluster = mServer.getHazelcastInstance().getCluster();
Set<Member> members = cluster.getMembers();
List<String> collect = members.stream().map(member -> member.getStringAttribute(HZ_Cluster_Node_ID)).collect(Collectors.toList());
log.info("hazelcast node list:{}",JSON.toJSONString(collect));
Map<String, Member> memberMap = members.stream().collect(Collectors.toMap(
member -> member.getStringAttribute(HZ_Cluster_Node_ID), member -> member, (k1, k2 )->k1));
//按照节点分发
nodeMap.forEach((nodeId, set) -> {
// 转发到其他节点发送
if (!nodeId.equals(localNodeId) && memberMap.containsKey(nodeId)) {
WFCMessage.NotifyMessage2Receivers notifyMessage2Receivers = WFCMessage.NotifyMessage2Receivers.newBuilder()
.setMessageId(messageId)
.addAllReceivers(set)
.setExceptClientId(exceptClientId==null?"":exceptClientId)
.setPullType(pullType)
.setTopic(topic)
.build();
clusterQueueService.publishMessage(nodeId,nodeId,null, IMTopic.NotifyMessage2ReceiversTopic, notifyMessage2Receivers.toByteArray());
}
// 当前节点处理发送
else {
WFCMessage.Message message = mServer.getStore().messagesStore().getMessage(messageId);
if (message != null) {
// Add By Youqibin 16:11 2022/3/15 接收通知前置处理
preHandle(message, set);
mServer.getImBusinessScheduler().execute(() ->messagesPublisher.publish2Receivers(message, set, exceptClientId, pullType, localNodeId));
// Add By Youqibin 16:11 2022/3/15 接收通知后置处理
postHandle(message, set);
}
}
});
}
关键点解析：
- clusterQueueService.publishMessage, 使用Disruptor发送消息事件，高性能异步处理
2.3 事件处理器 onEvent
当 Disruptor 事件发布后，ClusterPublishEventHandler.onEvent 负责实际的消息转发逻辑。
public class ClusterPublishEventHandler implements EventHandler<ClusterPublishEvent> {
private final Server server;
private final String nodeId;
public ClusterPublishEventHandler(Server server, String nodeId) {
this.server = server;
this.nodeId = nodeId;
}
@Override
public void onEvent(ClusterPublishEvent event, long sequence, boolean endOfBatch) {
log.info("Processing event: {} on node: {}", event, nodeId);
server.forwardMessage(nodeId, event.getFromUser(), event.getClientId(), event.getTopic(), event.getPayload());
}
}
关键点解析：
-
onEvent方法接收到ClusterPublishEvent后，调用server.forwardMessage进行消息转发。 -
endOfBatch用于标识当前事件是否为批处理中的最后一个事件。 -
log.info记录消息处理的关键日志，便于后续排查。
3. 总结
本文介绍了 Disruptor 在 IM 系统中的应用，核心逻辑包括：
-
初始化 Disruptor：为每个
nodeId创建独立的 RingBuffer。 - 按照节点转发消息：将用户消息存入对应节点的 RingBuffer。
-
消息处理：
onEvent方法从 RingBuffer 读取消息，并执行转发。
通过 Disruptor，可以大幅降低锁竞争，提高 IM 系统的吞吐量，使其能够在高并发环境下稳定运行。
4. 最后
欢迎关注加瓦点灯，每天推送干货知识，一起进步！
本文由mdnice多平台发布
