使用ActiveMQ实现阶梯式消息通知

开始先分享一下借鉴的帖子有：
1.Spring整合JMS(二)——三种消息监听器
2.RabbitMQ 延迟队列实现订单支付结果异步阶梯性通知
推荐先把上面两个文章看完，会对后面的阅读有帮助。

消息队列

消息队列有很多种：ActiveMQ，RabbitMQ，ZeroMQ，Kafka，RocketMQ等。本文不再细述AMQ是什么、如何搭建使用，初学者可先学习AMQ原理和用法。
翻了很多帖子很少有用AMQ实现梯度通知的，所以特此写一个。

应用场景

一个的交易系统，当用户消费成功后，需要发送通知，告知这笔订单的商户，该订单已被支付了。这种业务场景与微信和支付宝的支付成功回调通知雷同。微信支付结果通知接口文档

代码实现

业务代码：
代码用了lombok，所以对 @Data 这个注解陌生的同学先导入相应的jar包才能使用。以下是几个简单的POJO类。

/*** @Author: Gavin* @DateTime: 2018/8/14 * @Des: 消费的请求类，一些敏感字段我就不展示了*/
@Data
public class SaleReq extends BaseReq {//商户号private String merchId;//金额private String totallAmt;//卡号private String ecardId;//订单号private String orderId;//回调通知地址private String notifyUrl;private String appId;//订单的附属数据，可以理解为备注remarkprivate String attach;//支付通知发送次数private int notifyTime;
}

/*** @Author: Gavin* @DateTime: 2018/8/14 * @Des: 回调通知请求类*/
@Data
public class Notify4ConsumReq implements Serializable {private String appId;private String timestamp;private String orderId;private String amount;private String ecardId;private String storeId;private String attach;private String resultCode;private String errCode;private String errorMsg;private String sign;
}


/*** @Author: Gavin* @DateTime: 2018/8/14* @Des:*/
@Data
public class Notify4ConsumRsp implements Serializable {//响应码private String retCode;//响应信息private String retMsg;
}

消费业务执行完后判断消费是否成功，成功并且请求带有支付成功回调通知url和appid则调用回调通知方法，产生一个通知消息。
PS：appid是用于报文验签，防止第三方商户收到假的支付成功回调通知。

/*** @Author: Gavin* @DateTime: 2018/8/14 * @Des: 消费的controller*/
@Controller
public class SaleController{@AutowiredISaleService mSaleService;@AutowiredISaleAsynService saleAsynService;@RequestMapping(value = "/Sale", method = RequestMethod.POST, produces = {"application/json;charset=UTF-8"})@ResponseBodypublic String Sale(@RequestBody CommunicationReq param) {try {//消费业务，具体代码忽略rsp = mSaleService.doSale(procid, req, rsp);} catch (Exception e) {rsp.setRetCode(Constant.FAIL);rsp.setRetMsg("系统错误");return rsp;}//支付回调，当消费成功 && 有回调地址 && 请求带有appid 则调用支付回调方法（消息队列的生产者）if(Constant.SUCCESS.equals(rsp.getRetCode()) && StringUtils.isNotBlank(req.getNotifyUrl()) && StringUtils.isNotBlank(req.getAppId())){try {saleAsynService.notify4Produce(procid, req);}catch (Exception e){}}return rsp;}
}

/*** @Author: Gavin* @DateTime: 2018/8/14 * @Des: 消费的异步处理业务*/
@Service("saleAsynService")
public class SaleAsynServiceImpl implements ISaleAsynService {@AutowiredProducerService producerService;@Autowired@Qualifier("adapterQueue")Destination adapterQueue;@AutowiredIAppInfoDao appInfoDao;@Overridepublic void notify4Produce(String procid, SaleReq saleReq) {//产生消息log.info("发送支付通知内容：" + JsonUtil.Object2Json(saleReq));//初始发送次数为第一次 这是一个关键点saleReq.setNotifyTime(1);producerService.sendMessage(adapterQueue, saleReq, 0);log.info("发送支付通知成功！！！");}@Transactional(readOnly = true)@Overridepublic Notify4ConsumRsp notify4Consum(String procid, SaleReq saleReq) {Notify4ConsumRsp rsp = new Notify4ConsumRsp();rsp.setRetCode("success");//查找appid//找不到appid，无法做签名，所以直接不发通知。if(StringUtils.isBlank(saleReq.getAppId())){log.info("无appid1，不发送通知");return rsp;}AppInfo appInfo = appInfoDao.get(saleReq.getAppId());if(appInfo==null){log.info("无appid2，不发送通知");return rsp;}//拼装通知请求对象Notify4ConsumReq req = new Notify4ConsumReq();req.setAppId(saleReq.getAppId());req.setAmount(saleReq.getTotallAmt());req.setAttach(saleReq.getAttach());req.setEcardId(saleReq.getEcardId());req.setOrderId(saleReq.getOrderId());req.setResultCode("success");req.setStoreId(saleReq.getMerchId());//时间戳是为了签名的不确定性，此字段可以换成一个随机字符串req.setTimestamp(DateUtil.formatDateToString(new Date(),"yyyyMMddHHmmsss"));//生成签名String sign="";try {sign = SignUtil.generateSignature(SignUtil.beanToMap(req),appInfo.getAppSecret(),SignType.valueOf(appInfo.getSignType()));} catch (Exception e) {log.info("签名失败="+CommonUtil.getTrace(e));}req.setSign(sign);//向第三方支付系统发送校验请求String str=null;if(saleReq.getNotifyUrl().contains("http")){str = HttpClientUtil.postHttpJson(saleReq.getNotifyUrl() , JsonUtil.Object2Json(req), 10*1000);}else{str = HttpClientUtil.postHttpsJson(saleReq.getNotifyUrl() , JsonUtil.Object2Json(req), Constant.TIME_OUT);}if (str == null) {log.info("请求支付通知系统失败，数据返回为空");rsp.setRetCode("fail");return rsp;}log.info("支付通知系统返回结果：" + str);rsp = (Notify4ConsumRsp) JsonUtil.json2Object(str,Notify4ConsumRsp.class);return rsp;}
}

上面的代码都是和业务有关，以下是AMQ的代码实现了。

/*** @Author: Gavin* @DateTime: 2018/8/14 * @Des: 消息生产者*/
@Component
public class ProducerServiceImpl implements ProducerService {@Autowiredprivate JmsTemplate jmsTemplate;@Autowired@Qualifier("responseQueue")private Destination responseDestination;@Overridepublic void sendMessage(String procid,Destination destination, final String message,final int delayTime) {log.info("---------------生产者开始发送消息-----------------");log.info("消息内容：" + message);jmsTemplate.send(destination, new MessageCreator() {@Overridepublic Message createMessage(Session session) throws JMSException {TextMessage textMessage = session.createTextMessage(message);textMessage.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY,delayTime);return textMessage;}});log.info("---------------生产者发送消息完毕-----------------");}@Overridepublic void sendMessage(final Destination destination, final Serializable obj,final long delayTime) {jmsTemplate.convertAndSend(destination, obj, new MessagePostProcessor() {@Overridepublic Message postProcessMessage(Message message) throws JMSException {message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, delayTime);return message;}});}
}

/*** @Author: Gavin* @DateTime: 2018/8/14 * @Des: 消息消费者，监听接收消息*/
public class ConsumerMessageListener implements SessionAwareMessageListener {@Autowired@Qualifier("myMessageConverter")private MessageConverter messageConverter;@Autowired@Qualifier("adapterQueue")private Destination adapterQueue;@AutowiredISaleAsynService saleAsynService;@AutowiredProducerService producerService;private static final Map<Integer, Long> myMap;//初始话一个map，这个map的key代表通知重发次数，value是通知间隔时间//比如，第一次重发间隔5秒，第二次重发间隔10秒以此类推//通知频率为5/10/30/180/1800，单位：秒static {myMap = new HashMap<Integer, Long>();myMap.put(1, 5 * 1000L);myMap.put(2, 10 * 1000L);myMap.put(3, 30 * 1000L);myMap.put(4, 180 * 1000L);myMap.put(5, 1800 * 1000L);
//        myMap.put(6, 1800*1000L);
//        myMap.put(7, 1800*1000L);
//        myMap.put(8, 3600*1000L);}@Overridepublic void onMessage(Message message, Session session) {try {if  (message instanceof ObjectMessage) {ObjectMessage objMessage = (ObjectMessage) message;Object obj = messageConverter.fromMessage(objMessage);log.info("接收到一个ObjectMessage：" + JsonUtil.Object2Json(obj));if (obj instanceof SaleReq) {SaleReq saleReq = (SaleReq) obj;log.info("这是一个SaleReq类");Notify4ConsumRsp rsp = saleAsynService.notify4Consum(procid, saleReq);log.info("Notify4ConsumRsp=" + JsonUtil.Object2Json(rsp));if ("success".equals(rsp.getRetCode())) {//通知成功} else {//通知失败，重发//重发一次，通知次数+1，当次数超过预设的值5时，就不再重发通知。int notifytime = saleReq.getNotifyTime();LogUtil.addProclog(procid, "Time=" + notifytime);if (notifytime <= 5) {saleReq.setNotifyTime(saleReq.getNotifyTime() + 1);producerService.sendMessage(adapterQueue, saleReq, myMap.get(notifytime));}}} else {log.info("不能识别的JAVA类");}}} catch (Exception e) {log.info("消费者处理错误=" + CommonUtil.getTrace(e));}}}

/*** @Author: Gavin* @DateTime: 2018/8/14 * @Des: 这个是message的转换类。*/
public class MyMessageConverter implements MessageConverter {@Overridepublic Message toMessage(Object object, Session session)throws JMSException, MessageConversionException {return session.createObjectMessage((Serializable) object);}@Overridepublic Object fromMessage(Message message) throws JMSException,MessageConversionException {ObjectMessage objMessage = (ObjectMessage) message;return objMessage.getObject();}}

ActiveMQ的配置文件就不贴出来了，看完第一篇文章应该已经把ActiveMQ和spring融合一起了。
代码有很多删减过，所以直接复制代码是行不通的，仅展示主要的业务逻辑。

第二次写博，如有疑问或者建议可以给我留言哦，期待你的留言(●’◡’●)