返利公众号消息推送的可靠性保障：模板消息队列化与送达状态追踪
返利公众号消息推送的可靠性保障：模板消息队列化与送达状态追踪
大家好，我是 微赚淘客系统3.0 的研发者省赚客！
在微赚淘客系统3.0中，用户完成订单后需通过微信公众号模板消息实时通知返利到账。然而，直接调用微信接口存在网络抖动、限流、Token失效等问题，导致消息丢失或延迟。为保障高可靠推送，我们引入消息队列 + 异步重试 + 送达状态追踪机制，实现99.95%以上的消息成功送达率。
一、整体架构设计
系统采用三层架构：
- 消息生产层：业务逻辑触发消息入队；
- 消息处理层：消费队列，调用微信接口并记录状态；
- 状态追踪层：持久化发送结果，支持失败重试与人工干预。
核心依赖组件包括 RabbitMQ（或 RocketMQ）、Redis（用于 Token 缓存）、MySQL（消息状态表）。
二、消息实体与状态定义
首先定义消息模型：
packagejuwatech.cn.model;importjava.time.LocalDateTime;publicclassWechatTemplateMessage{privateLongid;privateStringopenId;// 用户 openidprivateStringtemplateId;// 模板 IDprivateStringpage;// 跳转小程序页面privateObjectdata;// 模板数据（JSON）privateIntegerretryCount;// 重试次数privateStringstatus;// PENDING, SENT, FAILED, EXPIREDprivateLocalDateTimecreateTime;privateLocalDateTimesendTime;// getters & setters}状态枚举：
PENDING：待发送SENT：已成功发送FAILED：发送失败（可重试）EXPIRED：超过最大重试次数或超时（如7天未送达）
三、消息入队逻辑
在返利结算完成后，将消息写入数据库并投递到 MQ：
packagejuwatech.cn.service;importjuwatech.cn.model.WechatTemplateMessage;importjuwatech.cn.dao.MessageDao;importorg.springframework.amqp.rabbit.core.RabbitTemplate;importorg.springframework.stereotype.Service;@ServicepublicclassMessageProducerService{privatefinalMessageDaomessageDao;privatefinalRabbitTemplaterabbitTemplate;publicMessageProducerService(MessageDaomessageDao,RabbitTemplaterabbitTemplate){this.messageDao=messageDao;this.rabbitTemplate=rabbitTemplate;}publicvoidenqueueMessage(StringopenId,StringtemplateId,Objectdata){WechatTemplateMessagemsg=newWechatTemplateMessage();msg.setOpenId(openId);msg.setTemplateId(templateId);msg.setData(data);msg.setStatus("PENDING");msg.setRetryCount(0);msg.setCreateTime(LocalDateTime.now());Longid=messageDao.insert(msg);rabbitTemplate.convertAndSend("wechat.notify.exchange","template.send",id);}}四、消息消费与微信调用
消费者从队列取出消息ID，执行发送逻辑：
packagejuwatech.cn.consumer;importjuwatech.cn.model.WechatTemplateMessage;importjuwatech.cn.service.WechatApiService;importjuwatech.cn.dao.MessageDao;importorg.springframework.amqp.rabbit.annotation.RabbitListener;importorg.springframework.stereotype.Component;@ComponentpublicclassTemplateMessageConsumer{privatefinalMessageDaomessageDao;privatefinalWechatApiServicewechatApiService;publicTemplateMessageConsumer(MessageDaomessageDao,WechatApiServicewechatApiService){this.messageDao=messageDao;this.wechatApiService=wechatApiService;}@RabbitListener(queues="wechat.template.queue")publicvoidhandle(LongmessageId){WechatTemplateMessagemsg=messageDao.findById(messageId);if(msg==null||!"PENDING".equals(msg.getStatus()))return;try{booleansuccess=wechatApiService.sendTemplateMessage(msg.getOpenId(),msg.getTemplateId(),msg.getData(),msg.getPage());msg.setSendTime(LocalDateTime.now());if(success){msg.setStatus("SENT");}else{handleFailure(msg);}}catch(Exceptione){handleFailure(msg);}finally{messageDao.updateStatus(msg);}}privatevoidhandleFailure(WechatTemplateMessagemsg){intmaxRetry=5;if(msg.getRetryCount()>=maxRetry){msg.setStatus("EXPIRED");}else{msg.setRetryCount(msg.getRetryCount()+1);msg.setStatus("FAILED");// 延迟重试：1min, 5min, 15min, 1h, 24hlongdelay=calculateDelay(msg.getRetryCount());// 重新入队（带延迟）requeueWithDelay(msg.getId(),delay);}}privatelongcalculateDelay(intretry){long[]delays={60_000L,300_000L,900_000L,3_600_000L,86_400_000L};returndelays[Math.min(retry-1,delays.length-1)];}privatevoidrequeueWithDelay(Longid,longdelayMs){// 使用 RabbitMQ TTL + 死信队列 或 RocketMQ 延迟消息// 此处简化为定时任务扫描 FAILED 状态消息（生产环境建议用原生延迟队列）}}五、微信 API 封装与 Token 管理
微信 access_token 有效期2小时，需全局缓存并自动刷新：
packagejuwatech.cn.service;importcom.fasterxml.jackson.databind.JsonNode;importjuwatech.cn.util.HttpClientUtil;importorg.springframework.data.redis.core.StringRedisTemplate;importorg.springframework.stereotype.Service;importjavax.annotation.PostConstruct;importjava.util.concurrent.TimeUnit;@ServicepublicclassWechatApiService{privatestaticfinalStringTOKEN_KEY="wechat:access_token";privatestaticfinalStringAPP_ID="wx123456";privatestaticfinalStringSECRET="secret789";privatefinalStringRedisTemplateredisTemplate;publicWechatApiService(StringRedisTemplateredisTemplate){this.redisTemplate=redisTemplate;}@PostConstructpublicvoidpreloadToken(){refreshAccessToken();}privatevoidrefreshAccessToken(){Stringurl="https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&appid="+APP_ID+"&secret="+SECRET;JsonNoderesp=HttpClientUtil.getJson(url);if(resp.has("access_token")){Stringtoken=resp.get("access_token").asText();intexpires=resp.get("expires_in").asInt();redisTemplate.opsForValue().set(TOKEN_KEY,token,expires-300,TimeUnit.SECONDS);}}publicbooleansendTemplateMessage(StringopenId,StringtemplateId,Objectdata,Stringpage){Stringtoken=redisTemplate.opsForValue().get(TOKEN_KEY);if(token==null){refreshAccessToken();token=redisTemplate.opsForValue().get(TOKEN_KEY);if(token==null)returnfalse;}StringsendUrl="https://api.weixin.qq.com/cgi-bin/message/template/send?access_token="+token;Stringpayload=buildPayload(openId,templateId,data,page);JsonNoderesult=HttpClientUtil.postJson(sendUrl,payload);returnresult.has("errcode")&&result.get("errcode").asInt()==0;}privateStringbuildPayload(StringopenId,StringtemplateId,Objectdata,Stringpage){// 构造 JSON，略return"{...}";}}六、送达状态追踪与监控
所有消息状态写入wechat_message_log表，支持以下能力：
- 定时任务扫描
FAILED消息进行补偿； - 管理后台查看失败原因（如“用户取消关注”、“模板ID无效”）；
- 埋点上报 Prometheus，监控推送成功率。
关键 SQL 示例：
-- 查询近1小时失败消息SELECT*FROMwechat_message_logWHEREstatus='FAILED'ANDcreate_time>NOW()-INTERVAL1HOUR;七、异常场景处理
- 用户取关：微信返回
{"errcode":43004}，标记为EXPIRED不再重试； - 模板被删除：返回
40003，告警通知运营人员； - 限流：返回
45009，立即暂停发送，10分钟后恢复。
通过上述机制，系统在日均50万条模板消息推送下，保持极低的失败率与人工干预成本。
本文著作权归 微赚淘客系统3.0 研发团队，转载请注明出处！