【即使通信 IM】如何保障消息的实时性、可靠性、有序性、幂等性_如何保证消息的实时性如何保证消息的时序性

关于即使通信 IM的开发，可以参考腾讯云即时通信IM。我们可以参考它的服务端和客户端API，它也为我们提供了一些问题的解决方案和设计思想。但如何保障消息的实时性、可靠性、幂等性、一致性需要根据具体开发代码而定，它并没有提供处理方案。这篇博客的话就说说如何保障消息的这些特性吧~

一、消息的实时性

消息的实时性主要是为了用户的体验。

一条消息发送给服务端，服务端快速处理完把消息结果给客户端，也可以说分发给消息接收方。如何让消息快速的让接收方都收到就是这里说的实时性。

一条消息发送给服务端，服务端需要经过以下处理：

• 验证消息（比如对方是否已被拉黑什么的，这个根据具体业务可有可无）
• 消息的存储
• 给发送消息者一个ACK相应，表示通过验证，发送成功
• 把消息发送给接收方，如果支持多端的话，还得保障多端同步
• 验证失败的话，得返回一个失败的ACK

接下来就说说如何保障消息的实时性的，没有绝对的实时，只能尽量优化。

一共三处优化吧：

1. 通信服务和业务处理服务一般是分开的，通信服务一般拿到消息都是交给业务服务进行处理。为提升用户体验，可以把验证消息放在通信服务层进行处理，然后失败了即时响应，成功了再去业务层处理剩余的业务。
   消息验证是放在业务层处理的，因为它与业务层的一些类有关联。如何处理呢?使用 Open Feign 去处理远程调用。如下面：

public interface FeignMessageService {

    @Headers({"Content-Type: application/json","Accept: application/json"})
    @RequestLine("POST /message/checkSend")
    ResponseVO checkSendMessage(CheckSendMessageReq o);

}

// TODO 1. 调用校验消息发送方的接口.
ResponseVO responseVO = feignMessageService.checkSendMessage(req);
// 如果成功投递到 mq，交给业务服务去处理
// 去存储，去分发
if (responseVO.isOk()) {
       MqMessageProducer.sendMessage(message, command);
}
// 失败则直接ack

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

2. 利用线程池去处理校验后的处理，因为这些操作是串行的，而且不影响原本程序，所以交给其他后台线程处理在合适不过了也是一种优化。核心代码如下：

    private final ThreadPoolExecutor threadPoolExecutor;

    {

        AtomicInteger num = new AtomicInteger(0);

        threadPoolExecutor = new ThreadPoolExecutor(
                6, 8, 60, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(),
                new ThreadFactory() {

                    @Override
                    public Thread newThread(Runnable r) {
                        Thread thread = new Thread(r);
                        thread.setName("message-process-thread-" + num.getAndIncrement());
                        thread.setDaemon(true);
                        return thread;
                    }
                }
        );
    }

			// 使用线程池处理校验后的处理
            threadPoolExecutor.execute(() -> {
                messageStoreService.storeP2PMessage(messageContent);
                // 1. 回 ack 给自己
                ack(messageContent, ResponseVO.successResponse());
                // 2. 发消息给同步在线端
                syncToSender(messageContent, new ClientInfo(messageContent.getAppId(), messageContent.getClientType(), messageContent.getImei()));
                // 3. 发消息给对方在线端
                dispatchMessage(messageContent);
            });
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

3. 由于消息的存储是IO操作，过程相比CPU去处理程序是很慢的，所以我们也应该想办法优化。如何优化呢？可以利用消息队列，把要处理的消息放入队列中，然后由监听这个队列的消费者进行处理存储，这过程是异步的，可以提升整个消息处理的速度。这边我用的是RabbitMQ，大伙可以看看下面代码：

    public void storeP2PMessage(MessageContent messageContent) {
        // TODO 发送消息
        rabbitTemplate.convertAndSend(Constants.RabbitConstants.StoreP2PMessage,
                "",
                JSONObject.toJSONString(dto));

    }


// 监听消息，然后消费消息，存储消息
@Service
@Slf4j
public class StoreP2PMessageReceiver {


    @Autowired
    StoreMessageService storeMessageService;

    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(value = Constants.RabbitConstants.StoreP2PMessage, durable = "true"),
                    exchange = @Exchange(value = Constants.RabbitConstants.StoreP2PMessage, durable = "true")
            ), concurrency = "1"
    )
    public void storeP2PMessage(@Payload Message message,
                                @Headers Map<String, Object> headers,
                                Channel channel) throws IOException {
        String msg = new String(message.getBody(), "utf-8");
        log.info("CHAT MSG FORM QUEUE ::: {}", msg);
        Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
        try {
            JSONObject jsonObject = JSON.parseObject(msg);
            DoStoreP2PMessageDto dto = jsonObject.toJavaObject(DoStoreP2PMessageDto.class);
            ImMessageBody messageBody = jsonObject.getObject("messageBodyDto", ImMessageBody.class);
            dto.setMessageBody(messageBody);
            storeMessageService.doStoreP2PMessage(dto);
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            log.error("处理消息出现异常：{}", e.getMessage());
            log.error("RMQ_CHAT_TRAN_ERROR", e);
            log.error("NACK_MSG:{}", msg);
            //第一个false 表示不批量拒绝，第二个false表示不重回队列
            channel.basicNack(deliveryTag, false, false);
        }
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

二、消息的可靠性

大伙都知道 TCP 本身就是具有可靠性的，但是它只能保障一应用到另一应用的传输层可靠，而应用层之间的可靠性并不能保证（比如消息到了传输层，然后应用闪退了，自然就没有到应用层处理消息，消息也自然没有存储，就丢失了可靠性）。通过 TCP 协议进行消息传输如下图所示：

IM 的开发本质是通过 im 服务去处理消息的，所以是需保障应用与 im 服务之间的可靠性。就如下图所示：

如何保障应用之间的可靠性呢?可靠性的保障就是让发送方知道接收方接收到了消息，这样就表示消息成功传递了。 如下图所示，即接收方需接收俩个ack才说明消息成功发送，一个是告诉发送方消息到达了 im 服务端且消息已被存储，一个是告诉发送方消息成功送达接收方。

接受方回 ack 有俩种，一种是接收方收到消息后让给发送方发送ack；另一种是接收方不在线，服务方发送发送方ack表示已经分发了消息。

• 接收方在线收到消息发送ack

	if (command == MessageCommand.MSG_RECIVE_ACK.getCommand()) {
                // 接收方收到消息回 ack
                MessageReciveAckContent messageContent = o.toJavaObject(MessageReciveAckContent.class);
                messageSyncService.receiveMark(messageContent);
            }

    @Autowired
    MessageProducer messageProducer;

    public void receiveMark(MessageReciveAckContent messageReciveAckContent){
        messageProducer.sendToUser(messageReciveAckContent.getToId(),
                MessageCommand.MSG_RECIVE_ACK,
                messageReciveAckContent,messageReciveAckContent.getAppId());
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14

• 接收方不在线，服务端回ack

// 3. 发消息给对方在线端
List<ClientInfo> clientInfos = dispatchMessage(messageContent);
					// 判断接收方是否有在线的
            if (clientInfos.isEmpty()) {
                revicerAck(messageContent);
            }
1
2
3
4
5
6

三、消息的有序性

由于在保障消息实时性的时候，用了线程池去处理消息的存储和分发，这就有可能导致多条消息发来导致乱序问题，就比如说俩条消息发来，前发的在后发的后面，这是由于多线程处理消息的分发，所以可能发送消息的乱序。

这里解决消息的有序性是利用了Redis原子性递增，每次消息到服务端都为其生成一个Redis原子性递增产生的消息序列号，然后交给前端，让前端根据序列号对消息进行排序。

        // 获取消息序列号 Seq，准备返回给前端，让前端处理消息的有序性
        long seq = redisSeq.doGetSeq(
                messageContent.getAppId() + ":"
                        + Constants.SeqConstants.Message + ":" + ConversationIdGenerate.generateP2PId(
                        messageContent.getFromId(), messageContent.getToId()
                )
        );
        messageContent.setMessageSequence(seq);


@Service
public class RedisSeq {

    @Autowired
    StringRedisTemplate stringRedisTemplate;

    public long doGetSeq(String key){
        return stringRedisTemplate.opsForValue().increment(key);
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

四、消息的幂等性

由于去保障了消息的可靠性，前端在一段时间内没收到俩个 ack 会重新发送这个消息，从而可能导致消息的幂等性。比如说发送了一条消息，存储了但是由于某些原因超时或者没有分发成功，然后前端又发了同一个消息（即messageId相同），导致存储了俩次，导致也可能分发了俩次，从而消息不再幂等。

解决方法：因为消息id的唯一，我们可以根据消息id进行去重，即用Redis缓存，当一条消息存储后也根据消息id在Redis中进行缓存，然后发同消息处理时，先判断缓存中是否有这条消息，如果有则直接分发操作即可。

    public void process(MessageContent messageContent) {

        MessageContent cache = messageStoreService.getMessageFromMessageIdCache(messageContent.getAppId(),
                messageContent.getMessageId(),
                MessageContent.class);
        // 如果消息已经被存储过了，那直接进行分发即可
        // 即不需要再次进行存储
        if (!ObjectUtils.isEmpty(cache)) {
            threadPoolExecutor.execute(() -> {
                // 1. 回 ack 给自己
                ack(messageContent, ResponseVO.successResponse());
                // 2. 发消息给同步在线端
                syncToSender(cache, new ClientInfo(cache.getAppId(), cache.getClientType(), cache.getImei()));
                // 3. 发消息给对方在线端
                List<ClientInfo> clientInfos = dispatchMessage(cache);
                if (clientInfos.isEmpty()) {
                    revicerAck(cache);
                }
            });
            return;
        }

        // 获取消息序列号 Seq，准备返回给前端，让前端处理消息的有序性
        long seq = redisSeq.doGetSeq(
                messageContent.getAppId() + ":"
                        + Constants.SeqConstants.Message + ":" + ConversationIdGenerate.generateP2PId(
                        messageContent.getFromId(), messageContent.getToId()
                )
        );
        messageContent.setMessageSequence(seq);

        // 前置校验
        // 这个用户是否被禁言、是否被禁用
        // 发送方和接收方是否是好友
        // ResponseVO responseVO = imServerPermissionCheck(fromId, toId, messageContent.getAppId());
        // if (responseVO.isOk()) {
        // 使用线程池处理校验后的处理
        threadPoolExecutor.execute(() -> {
            messageStoreService.storeP2PMessage(messageContent);
            // 1. 回 ack 给自己
            ack(messageContent, ResponseVO.successResponse());
            // 2. 发消息给同步在线端
            syncToSender(messageContent, new ClientInfo(messageContent.getAppId(), messageContent.getClientType(), messageContent.getImei()));
            // 3. 发消息给对方在线端
            List<ClientInfo> clientInfos = dispatchMessage(messageContent);

            messageStoreService.setMessageFromMessageIdCache(messageContent.getAppId(),
                    messageContent.getMessageId(),
                    messageContent);

            if (clientInfos.isEmpty()) {
                revicerAck(messageContent);
            }
        });

    }

    public void setMessageFromMessageIdCache(Integer appId, String messageId,
                                             Object messageContent) {
        String key = appId + ":" + Constants.RedisConstants.cacheMessage + ":"
                + messageId;
        stringRedisTemplate.opsForValue().set(key, JSONObject.toJSONString(messageContent),
                300, TimeUnit.SECONDS);
    }

    public <T> T getMessageFromMessageIdCache(Integer appId, String messageId,
                                              Class<T> clazz) {
        String key = appId + ":" + Constants.RedisConstants.cacheMessage + ":"
                + messageId;
        String obj = stringRedisTemplate.opsForValue().get(key);
        return JSONObject.parseObject(obj, clazz);
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72

五、总结