activeMQ 了解一下(三)——重发机制+项目应用
2018-07-19 本文已影响0人
夏日橘子冰
-
项目场景：
推送通知给第三方，并得到第三方的feedback，json交互
-
问题：
推送过程遇到网络异常？第三方异常？数据格式错误？等等问题怎么办
【这就是重试机制了，即什么时候触发+触发后做什么】
-
希望达成的目的：
1.网络异常，代码异常，对方响应码为失败等无关业务数据的推送失败，重试发送消息，上限为3次
2.数据异常，针对消息里数据解析后对方给回失败响应，不重发，记录发送失败
-
实现方式：
1，使用activemq，mq的好处见参见文章activeMQ了解一下(一)。且其有重发机制，刚好适合本场景（配置mq）
2，每次重发，都记录次数；视情况记录发送结果为失败/成功（数据库记录）
搭建mq见文章activeMQ了解一下(二)，本文只说如何配置重发机制
【第一步，配置xml】
spring-activemq.xml如下
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:tx="http://www.springframework.org/schema/tx"
xmlns:mongo="http://www.springframework.org/schema/data/mongo"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-3.2.xsd
http://www.springframework.org/schema/aop
http://www.springframework.org/schema/aop/spring-aop-3.2.xsd
http://www.springframework.org/schema/tx
http://www.springframework.org/schema/tx/spring-tx-3.2.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context-3.2.xsd">
<context:component-scan base-package="com.latech"/>
<!-- 重发机制 -->
<bean id="activeMQRedeliveryPolicy" class="org.apache.activemq.RedeliveryPolicy">
<!--是否在每次尝试重新发送失败后,增长这个等待时间 -->
<property name="useExponentialBackOff" value="true"></property>
<!--重发次数,默认为6次 这里设置为2次 -->
<property name="maximumRedeliveries" value="2"></property>
<!--重发时间间隔,默认为5秒，设置为1秒 -->
<property name="initialRedeliveryDelay" value="1000"></property>
<!--第一次失败后重新发送之前等待1秒,第二次失败再等待1 * 2秒,这里的2就是value -->
<property name="backOffMultiplier" value="2"></property>
<!--最大传送延迟,最大重发时间间隔时，以后每次重连时间间隔都为最大重连时间间隔。 -->
<property name="maximumRedeliveryDelay" value="10000"></property>
</bean>
<bean id="jmsFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">
<property name="connectionFactory">
<bean class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL"><value>tcp://192.168.28.2:61616?</value></property>
<property name="redeliveryPolicy" ref="activeMQRedeliveryPolicy" />
</bean>
</property>
<property name="maxConnections" value="100"></property>
</bean>
<!--使用缓存可以提升效率-->
<bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
<property name="targetConnectionFactory" ref="jmsFactory"/>
<property name="sessionCacheSize" value="100"/>
</bean>
<!-- 配置JMS模板（Queue），Spring提供的JMS工具类，它发送、接收消息。 -->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="connectionFactory"/>
<property name="messageConverter">
<bean class="org.springframework.jms.support.converter.SimpleMessageConverter"/>
</property>
</bean>
<!-- 定义推送中奖消息队列（Queue） -->
<bean id="awardMsgDestinationQueue" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg name="name" value="awardMsgDestinationQueue"/>
</bean>
<!-- 配置监听者（Queue） -->
<bean id="awardMsgQueueListener" class="com.latech.notify.consumer.AwardMsgQueueListener" />
<!-- 配置多个消息监听容器，配置连接工厂，监听的目标是defaultDestinationQueue，监听器是上面定义的监听器 -->
<bean id="queueListenerContainer1" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory" />
<property name="destination" ref="awardMsgDestinationQueue" />
<property name="messageListener" ref="awardMsgQueueListener" />
<property name="sessionTransacted" value="true"/>
</bean>
</beans>
敲重点：
- 要配置重发机制
- 监听器要开启session事务，见配置最后一行
【第二步，代码里触发】
public class AwardMsgQueueListener implements MessageListener{
private Logger logger = LoggerFactory.getLogger(this.getClass());
@Autowired
private IOrderService orderService;
@Override
@SuppressWarnings("unchecked")
public void onMessage(Message message) {
TextMessage tm = (TextMessage) message;
Integer sendTimes = 0;
String msg="",data = "",notifyUrl="",channel="";
try {
msg = tm.getText();
} catch (JMSException e1) {
logger.error("从消息队列获取消息出现异常，请检查");
e1.printStackTrace();
return;
}
logger.info("接收到的消息为："+msg);
try{
Map<String, Map<String, Object>> msgMap = JSONObject.parseObject(msg, new TypeReference<Map<String,Map<String,Object>>>(){});
notifyUrl = msgMap.keySet().iterator().next();
data = JSONObject.toJSONString(msgMap.get(notifyUrl));
// notifyUrl = "http://127.0.0.1:10003/order/receiveMsg.json";
Map<String, Object> dataMap = JSONObject.parseObject(data,new TypeReference<Map<String,Object>>(){});
channel = (String)dataMap.get("channel");
CloseableHttpClient httpClient = HttpClients.createDefault();
HttpPost httpPost= new HttpPost(notifyUrl);
httpPost.setHeader("Content-type", "application/json");
httpPost.setEntity(new StringEntity(data));
CloseableHttpResponse response = httpClient.execute(httpPost);
if(200 !=response.getStatusLine().getStatusCode()){
//请求失败时，记录推送次数,状态仍为推送中，不提交事务
logger.error("推送失败，statusCode为："+response.getStatusLine().getStatusCode());
this.updateOrderAfterFail(data);
throw new RuntimeException();
}else{
HttpEntity responseEntity = response.getEntity();
if(responseEntity == null || responseEntity.getContent()==null){
logger.error("推送商户中奖信息成功，但返回内容为空，url："+notifyUrl);
this.updateOrderAfterFail(data);
throw new RuntimeException("推送商户中奖信息成功，但返回内容为空，url："+notifyUrl);
}else{
StringBuilder entityStringBuilder = new StringBuilder();
BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(responseEntity.getContent(),"UTF-8"), 8 * 1024);
String line = null;
while ((line = bufferedReader.readLine()) != null) {
entityStringBuilder.append(line);
}
String result = entityStringBuilder.toString();
Map<String,Object> resultMap = JSONObject.parseObject(result,new TypeReference<Map<String,Object>>(){});
if(!resultMap.containsKey("data") || !resultMap.containsKey("security") || !resultMap.containsKey("response")){
logger.error("推送商户中奖信息成功，但返回内容不合规，响应数据中可能不包含“data,security,reponse”中的一个或多个");
this.updateOrderAfterFail(data);
throw new RuntimeException("推送商户中奖信息成功，但返回内容不合规，响应数据中可能不包含“data,security,reponse”中的一个或多个");
}
if("11111".equals((String)resultMap.get("response"))){
logger.error("推送商户中奖信息成功，但返回内容响应码为11111");
this.updateOrderAfterFail(data);
throw new RuntimeException("推送商户中奖信息成功，但返回内容响应码为11111");
}
List<Map<String,String>> orderList = (List<Map<String,String>>)resultMap.get("data");
for(Map<String,String> order : orderList){
List<OrderInfo> orderResult = orderService.getByOrderNumAndChannel(order.get("orderNum"),channel);
String serialNumber = orderResult.get(0).getSerialNumber();
OrderInfoVo existOrder = orderService.queryBySerialNumber(serialNumber);
OrderInfo orderInfo = new OrderInfo();
orderInfo.setSerialNumber(serialNumber);
orderInfo.setSendAwardTimes(existOrder.getSendAwardTimes()+1);
orderInfo.setSendAwardFlag(SendAwardFlagEnum.SEND_SUCCESS.getCode());
if("111111".equals(order.get("code"))){
orderInfo.setSendAwardFlag(SendAwardFlagEnum.SEND_FAIL.getCode());
}
orderService.updateByOrderInfo(orderInfo);
}
}
}
} catch (Exception e){
logger.error("推送消息异常捕捉："+e.getMessage());
e.printStackTrace();
this.updateOrderAfterFail(data);
throw new RuntimeException();//抛出此异常，触发重发机制
}
}
敲重点：
在需要重发时，抛出RuntimeException异常，会自动触发重发机制
由于还涉及到其他可能的异常，所以整体代码try..catch..最后统一throw
不过呢，现在还存在一个问题，配置里设置了重发次数为2，即包括初次发，一共发三次。但是最后数据库显示是最后发了6次的，6次是mq默认的次数，说明那个配置没生效，暂时还未解决这个问题
另外，消息队列里的消息这里是同步发送的，即按顺序一条条的来。前一条发送失败并重发时候，后一条是等待的。也可以配置异步发送，这里暂不做研究
PS：这里用到的消费者监听器时最简单基础的，还有一种SessionAwareMessageListener，重发机制的配置会略有不同，待尝试
