腾讯CMQ消息处理

CMQ延迟消息处理方式

针对延迟消息的时间满足不了需求的情况 , 把用户冻结的过期时间进行处理

用户的冻结时间未到期把消息进行重复投递回队列中 , 一直未过期就一直投递 , 直到用户冻结时间一过期把mysql库中的用户状态改为认证通过的状态

取消redis缓存作为判断冻结时间到期的依据, 自定义的消息体中增加到期时间字段 , 这样在消费的时候就知道其冻结的到期时间是多少, 如果当前消费的时间是在过期时间之前那么重复投递消息, 重复投递又有2个情况, 时间间隔超过1小时,按照1小时的过期时间投递, 如果不超过1小时的时间按照时间差进行投递, 如果是到了过期时间或者过期时间之后, 那么修改数据库的用户冻结状态, 发送解冻的短信给用户

package com.tencent.iov.userservice.config; import lombok.Data; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.cloud.context.config.annotation.RefreshScope; import org.springframework.stereotype.Component; import java.time.Duration; / * @author HP * @ Author; wangfei * @ Date ; 2021/5/8 17;00 * @ Version; 1.0 */ @Data @Component @ConfigurationProperties(prefix = "cmq") @RefreshScope public class CmqConfig { 
    / * 腾讯CMQ基本配置 */ private HttpConf httpConf; / * 用户冻结队列 */ private FrozenDelayQueueSetting frozenDelayQueueSetting; @Data public static class HttpConf { 
    private String secretId; private String secretKey; private String endpoint; private String path; private String method; } @Data public static class FrozenDelayQueueSetting { 
    private String queueName; private int consumerThreadCount; private Duration delayTime; } } 

基础功能类(基类)

package com.tencent.iov.userservice.cmq; import com.ruqimobility.common.dtos.BaseResponse; import com.ruqimobility.user.enummeta.UserResultEnum; import com.tencent.iov.parent.cmq.Account; import com.tencent.iov.parent.cmq.Queue; import com.tencent.iov.parent.cmq.Subscription; import com.tencent.iov.parent.cmq.Topic; import com.tencent.iov.userservice.config.CmqConfig; import com.tencent.iov.userservice.util.ResponseUtils; import lombok.extern.slf4j.Slf4j; import org.apache.commons.collections.CollectionUtils; import org.springframework.context.annotation.DependsOn; import org.springframework.stereotype.Service; import javax.annotation.Resource; import java.util.List; / * @author HP * @ Author: wangfei * @ Date : 2021/5/8 16:56 * @ Version: 1.0 */ @Slf4j @DependsOn(value = { 
   "cmqConfig"}) @Service public class BaseMqService { 
    @Resource private CmqConfig cmqConfig; public BaseResponse<Boolean> delMsgPerReq(String queueRealName, String receiptHandle) { 
    Account account = createAccount(); Queue queue = account.getQueue(queueRealName); if (null == queue) { 
    return ResponseUtils.fail(UserResultEnum.SERVICE_INNER_ERROR); } try { 
    queue.deleteMessage(receiptHandle); return ResponseUtils.success(Boolean.TRUE); } catch (Exception e) { 
    e.printStackTrace(); log.error("del msg fail. queue:" + queueRealName + " | receiptHandle:" + receiptHandle + " | e:{}", e); return ResponseUtils.fail(UserResultEnum.SERVICE_INNER_ERROR); } } / * 批量删除消息 * * @param queueRealName * @param receiptHandleList * @return */ public BaseResponse<Boolean> batchDelMsgPerReq(String queueRealName, List<String> receiptHandleList) { 
    if (CollectionUtils.isEmpty(receiptHandleList)) { 
    return ResponseUtils.fail(UserResultEnum.SERVICE_INNER_ERROR); } Account account = createAccount(); Queue queue = account.getQueue(queueRealName); if (null == queue) { 
    return ResponseUtils.fail(UserResultEnum.SERVICE_INNER_ERROR); } try { 
    queue.batchDeleteMessage(receiptHandleList); return ResponseUtils.success(Boolean.TRUE); } catch (Exception e) { 
    e.printStackTrace(); log.error("del msg fail. queue:" + queueRealName + " | e:{}", e); return ResponseUtils.fail(UserResultEnum.SERVICE_INNER_ERROR); } } protected Account createAccount() { 
    return new Account(cmqConfig.getHttpConf().getSecretId(), cmqConfig.getHttpConf().getSecretKey(), cmqConfig.getHttpConf().getEndpoint(), cmqConfig.getHttpConf().getPath(), cmqConfig.getHttpConf().getMethod()); } public CmqConfig.FrozenDelayQueueSetting getFrozenDelayQueueSetting() { 
    return cmqConfig.getFrozenDelayQueueSetting(); } public void createSubscribe(String topicName, String subscriptionName, String Endpoint, String Protocal) throws Exception { 
    createAccount().createSubscribe(topicName, subscriptionName, Endpoint, Protocal); } public Queue getQueue(String queue) { 
    return createAccount().getQueue(queue); } public Topic getTopic(String topicName) { 
    return createAccount().getTopic(topicName); } public Subscription getSubscription(String topicName, String subscriptionName) { 
    return createAccount().getSubscription(topicName, subscriptionName); } } 

cmq消费类

package com.tencent.iov.userservice.cmq; import com.alibaba.fastjson.JSON; import com.ruqimobility.common.dtos.BaseResponse; import com.ruqimobility.user.enummeta.UserResultEnum; import com.tencent.iov.parent.cmq.Account; import com.tencent.iov.parent.cmq.CMQServerException; import com.tencent.iov.parent.cmq.Message; import com.tencent.iov.parent.cmq.Queue; import com.tencent.iov.userservice.util.ResponseUtils; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; import java.util.Collections; import java.util.List; import java.util.Optional; / * @ Author: wangfei * @ Date : 2021/5/10 16:42 * @ Version: 1.0 */ @Slf4j @Service public class CmqConsumerService extends BaseMqService { 
    public Optional<Message> consumerMsg(String queueName) { 
    Queue queue = super.getQueue(queueName); if (null == queue) { 
    return Optional.empty(); } Message message = null; try { 
    message = queue.receiveMessage(); return Optional.ofNullable(message); } catch (CMQServerException e) { 
    return Optional.empty(); } catch (Exception e1) { 
    log.error("consume cmq msg fail. queue:" + queueName + " | message:" + JSON.toJSONString(message) + " | e:{}", e1); return Optional.empty(); } } / * 每次请求都new一个新的连接 * * @param queueRealName //真实的队列名称,不是spring bean 的名称 * @return */ public BaseResponse<Message> consumerMsgPerReq(String queueRealName) { 
    Account account = super.createAccount(); Queue queue = account.getQueue(queueRealName); if (null == queue) { 
    return ResponseUtils.fail(UserResultEnum.SERVICE_INNER_ERROR); } Message message = null; try { 
    message = queue.receiveMessage(); return ResponseUtils.success(message); } catch (CMQServerException e1) { 
    return ResponseUtils.fail(UserResultEnum.SERVICE_INNER_ERROR); } catch (Exception e) { 
    e.printStackTrace(); log.error("consume msg fail. queue:" + queueRealName + " | message:" + JSON.toJSONString(message) + " | e:{}", e); return ResponseUtils.fail(UserResultEnum.SERVICE_INNER_ERROR); } } / * 批量拉取消息,每批次最多16条,SDK规定 * * @param queueRealName * @param numOfBatchMsgPerPolling * @return */ public BaseResponse<List<Message>> batchConsumerMsgPerReq(String queueRealName, int numOfBatchMsgPerPolling) { 
    if (numOfBatchMsgPerPolling > 16) { 
    log.info("error"); } Account account = super.createAccount(); Queue queue = account.getQueue(queueRealName); if (null == queue) { 
    return ResponseUtils.fail(UserResultEnum.SERVICE_INNER_ERROR); } List<Message> messageList = Collections.emptyList(); try { 
    messageList = queue.batchReceiveMessage(numOfBatchMsgPerPolling); } catch (CMQServerException e1) { 
    return ResponseUtils.fail(UserResultEnum.SERVICE_INNER_ERROR); } catch (Exception e) { 
    log.error("consume msg fail. queue:" + queueRealName + " | message:" + JSON.toJSONString(messageList) + " | e:{}", e); return ResponseUtils.fail(UserResultEnum.SERVICE_INNER_ERROR); } return ResponseUtils.success(messageList); } } 

cmq生产者类

package com.tencent.iov.userservice.cmq; import com.ruqimobility.common.dtos.BaseResponse; import com.ruqimobility.user.enummeta.UserResultEnum; import com.tencent.iov.parent.cmq.Account; import com.tencent.iov.parent.cmq.CMQServerException; import com.tencent.iov.parent.cmq.Queue; import com.tencent.iov.userservice.util.ResponseUtils; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; import java.util.List; / * @author HP * @ Author: wangfei * @ Date : 2021/5/10 16:45 * @ Version: 1.0 */ @Slf4j @Service public class CmqProducerService extends BaseMqService { 
    public BaseResponse<Boolean> sendToQueue(String queueName, String msgBody, int delay) { 
    try { 
    Queue queue = super.getQueue(queueName); if (null == queue) { 
    return ResponseUtils.fail(UserResultEnum.SERVICE_INNER_ERROR); } if (delay > 0) { 
    queue.sendMessage(msgBody, delay); } else { 
    queue.sendMessage(msgBody); } return ResponseUtils.success(Boolean.TRUE); } catch (CMQServerException e1) { 
    log.error("CMQServerException. error:{}", e1); return ResponseUtils.fail(UserResultEnum.SERVICE_INNER_ERROR); } catch (Exception e) { 
    e.printStackTrace(); log.error("send msg fail. queue:" + queueName + " | msgBody:+" + msgBody + " | e:{}", e); return ResponseUtils.fail(UserResultEnum.SERVICE_INNER_ERROR); } } / * 每次发送消息都创建一次连接 * * @param queueRealName * @param msgBody * @param delay * @return */ public BaseResponse<Boolean> sendToQueuePerReq(String queueRealName, String msgBody, int delay) { 
    try { 
    Account account = super.createAccount(); Queue queue = account.getQueue(queueRealName); if (null == queue) { 
    return ResponseUtils.fail(UserResultEnum.SERVICE_INNER_ERROR); } if (delay > 0) { 
    queue.sendMessage(msgBody, delay); } else { 
    queue.sendMessage(msgBody); } return ResponseUtils.success(Boolean.TRUE); } catch (CMQServerException e1) { 
    log.error("CMQServerException. error:{}", e1); return ResponseUtils.fail(UserResultEnum.SERVICE_INNER_ERROR); } catch (Exception e) { 
    e.printStackTrace(); log.error("send msg fail. queue:" + queueRealName + " | msgBody:+" + msgBody + " | e:{}", e); return ResponseUtils.fail(UserResultEnum.SERVICE_INNER_ERROR); } } / * 批量发送消息 * * @param queueRealName * @param msgBody * @param delay * @return */ public BaseResponse<Boolean> batchSendToQueuePerReq(String queueRealName, List<String> msgBody, int delay) { 
    try { 
    Account account = super.createAccount(); Queue queue = account.getQueue(queueRealName); if (null == queue) { 
    return ResponseUtils.fail(UserResultEnum.SERVICE_INNER_ERROR); } if (delay > 0) { 
    queue.batchSendMessage(msgBody, delay); } else { 
    queue.batchSendMessage(msgBody); } return ResponseUtils.success(Boolean.TRUE); } catch (CMQServerException e1) { 
    log.error("CMQServerException. error:{}", e1); return ResponseUtils.fail(UserResultEnum.SERVICE_INNER_ERROR); } catch (Exception e) { 
    e.printStackTrace(); log.error("send msg fail. queue:" + queueRealName + " | msgBody:+" + msgBody + " | e:{}", e); return ResponseUtils.fail(UserResultEnum.SERVICE_INNER_ERROR); } } } 

冻结消费service

package com.tencent.iov.userservice.cmq; import com.alibaba.fastjson.JSON; import com.fasterxml.jackson.databind.ObjectMapper; import com.ruqimobility.common.dtos.BaseResponse; import com.ruqimobility.user.enummeta.SrAuthEnum; import com.ruqimobility.user.enummeta.UserRoleEnum; import com.tencent.iov.parent.cmq.Message; import com.tencent.iov.userservice.config.CmqConfig; import com.tencent.iov.userservice.dao.UserMapper; import com.tencent.iov.userservice.dto.message.UserRoleFrozenMsg; import com.tencent.iov.userservice.model.User; import com.tencent.iov.userservice.service.UserRoleFrozenService; import lombok.extern.slf4j.Slf4j; import org.springframework.boot.ApplicationArguments; import org.springframework.boot.ApplicationRunner; import org.springframework.stereotype.Component; import javax.annotation.PreDestroy; import javax.annotation.Resource; import java.io.IOException; import java.sql.Timestamp; import java.util.Optional; import java.util.concurrent.Executor; / * @author HP * @ Author: wangfei * @ Date : 2021/5/10 17:47 * @ Version: 1.0 */ @Slf4j @Component public class FrozenConsumerService extends CmqConsumerService implements ApplicationRunner { 
    private static boolean CONSUMER_RUNNING = true; @Resource(name = "taskAsyncPool") private Executor taskExecutor; @Resource private UserRoleFrozenService userRoleFrozenService; @Resource private FrozenCmqProducerService frozenCmqProducerService; @Resource private UserMapper userMapper; public void startUp() { 
    CmqConfig.FrozenDelayQueueSetting setting = super.getFrozenDelayQueueSetting(); // 队列名称 String queueName = setting.getQueueName(); // 消费线程数 int queueNum = setting.getConsumerThreadCount(); for (int i = 0; i < queueNum; i++) { 
    // 创建任务 UnFrozenUserTask unFrozenUserTask = new UnFrozenUserTask(queueName, this, userRoleFrozenService); // executor执行 taskExecutor.execute(unFrozenUserTask); } //注册钩子 Runtime.getRuntime().addShutdownHook(new Thread(() -> { 
    log.info("Order DispatchOrder The hook running..."); //第三步：调用停机处理 stop(); })); } @Override public void run(ApplicationArguments args) throws Exception { 
    log.info("========= consumer frozen queue message start ========= "); startUp(); } @PreDestroy public void stop() { 
    log.info("stop UnFrozenUser task consumer"); CONSUMER_RUNNING = false; } public class UnFrozenUserTask implements Runnable { 
    private String queueName; private FrozenConsumerService consumerService; private UserRoleFrozenService userRoleFrozenService; public UnFrozenUserTask(String queueName, FrozenConsumerService consumerService, UserRoleFrozenService userRoleFrozenService) { 
    this.queueName = queueName; this.consumerService = consumerService; this.userRoleFrozenService = userRoleFrozenService; } / * 执行任务 */ @Override public void run() { 
    // 消费解冻逻辑 while (CONSUMER_RUNNING) { 
    // 消费消息 Optional<Message> response = consumerService.consumerMsg(queueName); if (!response.isPresent()) { 
    try { 
    Thread.sleep(2000); } catch (InterruptedException e) { 
    log.error("consumer order status cmq thread sleep ex: {}", e); } // 消息为空continue 重复消费 continue; } //获取消息体 String msgBody = response.get().msgBody; log.info("get the userRoleFrozenMsg message body from cmq: {}", msgBody); try { 
    UserRoleFrozenMsg userRoleFrozenMsg = new ObjectMapper().readValue(msgBody, UserRoleFrozenMsg.class); User user = userMapper.getById(userRoleFrozenMsg.getUserId()); // 处于冻结状态时,判断是否冻结时间是否已过 if (user.getSrAuth() == SrAuthEnum.FROZEN.code) { 
    Timestamp now = new Timestamp(System.currentTimeMillis()); Timestamp endTime = userRoleFrozenMsg.getFrozenEndTime(); // 在结束时间之前继续投递 if (now.before(endTime)) { 
    long interval = (endTime.getTime() - now.getTime()) / 1000; // 间隔超过1小时时 , 按1小时投递延迟消息 if (interval > FrozenCmqProducerService.MAX_DELAY_SECOND) { 
    BaseResponse<Boolean> resp = frozenCmqProducerService.sendMessage(JSON.toJSONString(userRoleFrozenMsg)); log.info("Deliver messages more than 1 hour interval"); } else { 
    // 1小时内 , 按照时间差投递 BaseResponse<Boolean> baseResponse = frozenCmqProducerService.sendMessage(JSON.toJSONString(userRoleFrozenMsg), (int) interval); log.info("Deliver duplicate delivery messages within 1 hour"); } } else if (now.after(endTime)) { 
    // 在结束时间之后, 修改用户实名认证状态 userRoleFrozenService.autoUnFrozenUserRole(user.getUserId(), UserRoleEnum.SFC_USER); } } // 删除此次消费的消息 BaseResponse<Boolean> delMsgPerReq = consumerService.delMsgPerReq(queueName, response.get().receiptHandle); log.info("sr_auth equals 1 message delete userRoleFrozenMsg from cmq msg: {} response: {}", msgBody, delMsgPerReq); } catch (IOException e) { 
    log.error("userRoleFrozenMsg message:{} body parsing error:{}", msgBody, e); } } } } } 

冻结生产service

package com.tencent.iov.userservice.cmq; import com.ruqimobility.common.dtos.BaseResponse; import com.tencent.iov.userservice.config.CmqConfig; import com.tencent.iov.userservice.util.ResponseUtils; import org.springframework.stereotype.Service; import java.util.List; / * @author HP * @ Author: wangfei * @ Date : 2021/5/27 17:42 * @ Version: 1.0 */ @Service public class FrozenCmqProducerService extends CmqProducerService { 
    / * 最大延迟时间, 3600 秒 */ public static Integer MAX_DELAY_SECOND = 3600; / * 发送延迟消息 * * @param msgBody * @return */ public BaseResponse<Boolean> sendMessage(String msgBody) { 
    CmqConfig.FrozenDelayQueueSetting setting = super.getFrozenDelayQueueSetting(); Long seconds = setting.getDelayTime().getSeconds(); return super.sendToQueue(setting.getQueueName(), msgBody, seconds.intValue()); } / * 批量发送延迟冻结消息 * * @return */ public BaseResponse<Boolean> batchSendMessage(List<String> msgBodyList) { 
    CmqConfig.FrozenDelayQueueSetting setting = super.getFrozenDelayQueueSetting(); Long seconds = setting.getDelayTime().getSeconds(); return super.batchSendToQueuePerReq(setting.getQueueName(), msgBodyList, seconds.intValue()); } / * 发送自定义延迟时间消息 * * @param msgBody * @return */ public BaseResponse<Boolean> sendMessage(String msgBody, int delay) { 
    CmqConfig.FrozenDelayQueueSetting setting = super.getFrozenDelayQueueSetting(); if (delay > MAX_DELAY_SECOND) { 
    return ResponseUtils.success(Boolean.FALSE); } return super.sendToQueue(setting.getQueueName(), msgBody, delay); } / * 批量发送延迟冻结消息 * * @return */ public BaseResponse<Boolean> batchSendMessage(List<String> msgBodyList, int delay) { 
    CmqConfig.FrozenDelayQueueSetting setting = super.getFrozenDelayQueueSetting(); if (delay > MAX_DELAY_SECOND) { 
    return ResponseUtils.success(Boolean.FALSE); } return super.batchSendToQueuePerReq(setting.getQueueName(), msgBodyList, delay); } } 

// 删除此次消费的消息 BaseResponse<Boolean> delMsgPerReq = consumerService.delMsgPerReq(queueName, response.get().receiptHandle); log.info("sr_auth equals 1 message delete userRoleFrozenMsg from cmq msg: {} response: {}", msgBody, delMsgPerReq);