当前位置：首页 > article >正文

高性能队列 Disruptor 在 IM 系统中的实战

article 2025/1/24 8:36:30

高性能队列 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());
    }
}