赞
踩
关于即使通信 IM的开发,可以参考腾讯云即时通信IM。我们可以参考它的服务端和客户端API,它也为我们提供了一些问题的解决方案和设计思想。但如何保障消息的实时性、可靠性、幂等性、一致性需要根据具体开发代码而定,它并没有提供处理方案。这篇博客的话就说说如何保障消息的这些特性吧~
消息的实时性主要是为了用户的体验。
一条消息发送给服务端,服务端快速处理完把消息结果给客户端,也可以说分发给消息接收方。如何让消息快速的让接收方都收到就是这里说的实时性。
一条消息发送给服务端,服务端需要经过以下处理:
接下来就说说如何保障消息的实时性的,没有绝对的实时,只能尽量优化。
一共三处优化吧:
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
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);
});
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);
}
}
}
大伙都知道 TCP 本身就是具有可靠性的,但是它只能保障一应用到另一应用的传输层可靠,而应用层之间的可靠性并不能保证(比如消息到了传输层,然后应用闪退了,自然就没有到应用层处理消息,消息也自然没有存储,就丢失了可靠性)。通过 TCP 协议进行消息传输如下图所示:
IM 的开发本质是通过 im 服务去处理消息的,所以是需保障应用与 im 服务之间的可靠性。就如下图所示:
如何保障应用之间的可靠性呢?可靠性的保障就是让发送方知道接收方接收到了消息,这样就表示消息成功传递了。 如下图所示,即接收方需接收俩个ack才说明消息成功发送,一个是告诉发送方消息到达了 im 服务端且消息已被存储,一个是告诉发送方消息成功送达接收方。
接受方回 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());
}
// 3. 发消息给对方在线端
List<ClientInfo> clientInfos = dispatchMessage(messageContent);
// 判断接收方是否有在线的
if (clientInfos.isEmpty()) {
revicerAck(messageContent);
}
由于在保障消息实时性的时候,用了线程池去处理消息的存储和分发,这就有可能导致多条消息发来导致乱序问题,就比如说俩条消息发来,前发的在后发的后面,这是由于多线程处理消息的分发,所以可能发送消息的乱序。
这里解决消息的有序性是利用了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);
}
}
由于去保障了消息的可靠性,前端在一段时间内没收到俩个 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);
}
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。