核心内容摘要
Java处理JSON编程实用技巧
返利公众号消息推送的可靠性保障模板消息队列化与送达状态追踪大家好我是 微赚淘客系统
0 的研发者省赚客在微赚淘客系统
0中用户完成订单后需通过微信公众号模板消息实时通知返利到账。
然而直接调用微信接口存在网络抖动、限流、Token失效等问题导致消息丢失或延迟。
为保障高可靠推送我们引入消息队列 异步重试 送达状态追踪机制实现
9
95%以上的消息成功送达率。
整体架构设计系统采用三层架构消息生产层业务逻辑触发消息入队消息处理层消费队列调用微信接口并记录状态状态追踪层持久化发送结果支持失败重试与人工干预。
核心依赖组件包括 RabbitMQ或 RocketMQ、Redis用于 Token 缓存、MySQL消息状态表。
消息实体与状态定义首先定义消息模型packagejuwatech.cn.model;importjava.time.LocalDateTime;publicclassWechatTemplateMessage{privateLongid;privateStringopenId;// 用户 openidprivateStringtemplateId;// 模板 IDprivateStringpage;// 跳转小程序页面privateObjectdata;// 模板数据JSONprivateIntegerretryCount;// 重试次数privateStringstatus;// PENDING, SENT, FAILED, EXPIREDprivateLocalDateTimecreateTime;privateLocalDateTimesendTime;// getters setters}状态枚举PENDING待发送SENT已成功发送FAILED发送失败可重试EXPIRED超过最大重试次数或超时如7天未送达
消息入队逻辑在返利结算完成后将消息写入数据库并投递到 MQpackagejuwatech.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.messageDaomessageDao;this.rabbitTemplaterabbitTemplate;}publicvoidenqueueMessage(StringopenId,StringtemplateId,Objectdata){WechatTemplateMessagemsgnewWechatTemplateMessage();msg.setOpenId(openId);msg.setTemplateId(templateId);msg.setData(data);msg.setStatus(PENDING);msg.setRetryCount(
;msg.setCreateTime(LocalDateTime.now());LongidmessageDao.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.messageDaomessageDao;this.wechatApiServicewechatApiService;}RabbitListener(queueswechat.template.queue)publicvoidhandle(LongmessageId){WechatTemplateMessagemsgmessageDao.findById(messageId);if(msgnull||!PENDING.equals(msg.getStatus()))return;try{booleansuccesswechatApiService.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){intmaxRetry5;if(msg.getRetryCount()maxRetry){msg.setStatus(EXPIRED);}else{msg.setRetryCount(msg.getRetryCount()
;msg.setStatus(FAILED);// 延迟重试1min, 5min, 15min, 1h, 24hlongdelaycalculateDelay(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-
];}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_KEYwechat:access_token;privatestaticfinalStringAPP_IDwx123456;privatestaticfinalStringSECRETsecret789;privatefinalStringRedisTemplateredisTemplate;publicWechatApiService(StringRedisTemplateredisTemplate){this.redisTemplateredisTemplate;}PostConstructpublicvoidpreloadToken(){refreshAccessToken();}privatevoidrefreshAccessToken(){Stringurlhttps://api.weixin.qq.com/cgi-bin/token?grant_typeclient_credentialappidAPP_IDsecretSECRET;JsonNoderespHttpClientUtil.getJson(url);if(resp.has(access_token)){Stringtokenresp.get(access_token).asText();intexpiresresp.get(expires_in).asInt();redisTemplate.opsForValue().set(TOKEN_KEY,token,expires-300,TimeUnit.SECONDS);}}publicbooleansendTemplateMessage(StringopenId,StringtemplateId,Objectdata,Stringpage){StringtokenredisTemplate.opsForValue().get(TOKEN_KEY);if(tokennull){refreshAccessToken();tokenredisTemplate.opsForValue().get(TOKEN_KEY);if(tokennull)returnfalse;}StringsendUrlhttps://api.weixin.qq.com/cgi-bin/message/template/send?access_tokentoken;StringpayloadbuildPayload(openId,templateId,data,page);JsonNoderesultHttpClientUtil.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_logWHEREstatusFAILEDANDcreate_timeNOW()-INTERVAL1HOUR;
异常场景处理用户取关微信返回{errcode:43004}标记为EXPIRED不再重试模板被删除返回40003告警通知运营人员限流返回45009立即暂停发送10分钟后恢复。
通过上述机制系统在日均50万条模板消息推送下保持极低的失败率与人工干预成本。
本文著作权归 微赚淘客系统