【mq】从零开始实现 mq-10-消费者拉取消息回执 pull message ack

package com.github.houbb.mq.common.constant;  /**  * @author binbin.hou  * @since 0.0.3  */ public final class MessageStatusConst {      private MessageStatusConst(){}      /**      * 待消费      * ps: 生产者推送到 broker 的初始化状态      */     public static final String WAIT_CONSUMER = "W";      /**      * 推送给消费端处理中      * ps: broker 准备推送时，首先将状态更新为 P，等待推送结果      * @since 0.1.0      */     public static final String TO_CONSUMER_PROCESS = "TCP";      /**      * 推送给消费端成功      * @since 0.1.0      */     public static final String TO_CONSUMER_SUCCESS = "TCS";      /**      * 推送给消费端失败      * @since 0.1.0      */     public static final String TO_CONSUMER_FAILED = "TCF";      /**      * 消费完成      */     public static final String CONSUMER_SUCCESS = "CS";      /**      * 消费失败      */     public static final String CONSUMER_FAILED = "CF";      /**      * 稍后消费      * @since 0.1.0      */     public static final String CONSUMER_LATER = "CL";  }

消费者状态回执

我们在消费之后，添加状态回执：

for(MqMessage mqMessage : mqMessageList) {     IMqConsumerListenerContext context = new MqConsumerListenerContext();     final String messageId = mqMessage.getTraceId();     ConsumerStatus consumerStatus = mqListenerService.consumer(mqMessage, context);     log.info("消息：{} 消费结果 {}", messageId, consumerStatus);      // 状态同步更新     MqCommonResp ackResp = consumerBrokerService.consumerStatusAck(messageId, consumerStatus);     log.info("消息：{} 状态回执结果 {}", messageId, JSON.toJSON(ackResp)); }

回执实现，根据 messageId 更新对应的消息消费状态。

public MqCommonResp consumerStatusAck(String messageId, ConsumerStatus consumerStatus) {     final MqConsumerUpdateStatusReq req = new MqConsumerUpdateStatusReq();     req.setMessageId(messageId);     req.setMessageStatus(consumerStatus.getCode());     final String traceId = IdHelper.uuid32();     req.setTraceId(traceId);     req.setMethodType(MethodType.C_CONSUMER_STATUS);      // 重试     return Retryer.<MqCommonResp>newInstance()             .maxAttempt(consumerStatusMaxAttempt)             .callable(new Callable<MqCommonResp>() {                 @Override                 public MqCommonResp call() throws Exception {                     Channel channel = getChannel(null);                     MqCommonResp resp = callServer(channel, req, MqCommonResp.class);                     if(!MqCommonRespCode.SUCCESS.getCode().equals(resp.getRespCode())) {                         throw new MqException(ConsumerRespCode.CONSUMER_STATUS_ACK_FAILED);                     }                     return resp;                 }             }).retryCall(); }

Broker 回执处理

消息分发

// 消费者消费状态 ACK if(MethodType.C_CONSUMER_STATUS.equals(methodType)) {     MqConsumerUpdateStatusReq req = JSON.parseObject(json, MqConsumerUpdateStatusReq.class);     final String messageId = req.getMessageId();     final String messageStatus = req.getMessageStatus();     return mqBrokerPersist.updateStatus(messageId, messageStatus); }

简单实现

这里是基于本地 map 更新状态的，性能比较差。

后续会以 mysql 实现。

public MqCommonResp updateStatus(String messageId, String status) {     // 这里性能比较差，所以不可以用于生产。仅作为测试验证     for(List<MqMessagePersistPut> list : map.values()) {         for(MqMessagePersistPut put : list) {             MqMessage mqMessage = put.getMqMessage();             if(mqMessage.getTraceId().equals(messageId)) {                 put.setMessageStatus(status);                 break;             }         }     }      MqCommonResp commonResp = new MqCommonResp();     commonResp.setRespCode(MqCommonRespCode.SUCCESS.getCode());     commonResp.setRespMessage(MqCommonRespCode.SUCCESS.getMsg());     return commonResp; }