|
package cn.my.server.clientdemo.yuer;
import java.io.IOException;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.retry.RecoveryCallback;
import org.springframework.retry.RetryCallback;
import org.springframework.retry.RetryContext;
import org.springframework.retry.backoff.FixedBackOffPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.rabbitmq.client.Channel;
/**
* 余额宝端消息监听
*/
@Component
public class YuErBaoMessageListeners {
private final Logger log = LoggerFactory.getLogger(this.getClass());
private ObjectMapper mapper = new ObjectMapper();
/**
* 监听消息队列
* @param message 消息内容
* @param channel 消息渠道
* @throws IOException 异常
*/
@RabbitListener(queues = "money")
@RabbitHandler
public void receiveQueue(Message message, Channel channel) throws IOException {
String msg = "";
try {
// 业务处理逻辑
msg = new String(message.getBody());
Map data = mapper.readValue(msg, HashMap.class);
retry(data);
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);// 手动应答消息已经处理
} catch (Exception e) {
log.error("MQ接收消息内容[" + msg + "],后处理异常：" + e);
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);// 手动应答消息已经处理
}
}
/**
* 更新余额宝账户金额
*
* @param map
*/
@Transactional
public void bizOp(Map<String, Object> map) {
// B_MESSAGE交易流水表,主键是【交易编号】，重复插入会报异常，类似于幂等操作
System.out.println(
"【余额宝记账操作】重复插入会报异常，类似于幂等操作 insert into B_MESSAGE(transId,userId,money) values (000001,1,10000)");
// 更新余额宝账户金额
System.out.println("【余额宝账户入款】 update B set amount = amount +10000 where userId = 1");
}
/**
* 一直报错，重试次数用完了，保存如下信息供人工干预
*
* @param map
* @throws JsonProcessingException
*/
public void failed(Map<String, Object> map) throws JsonProcessingException {
System.out.println("一直报错，重试次数用完了，保存如下信息供人工干预:\n" + mapper.writeValueAsString(map));
}
/**
* 异常时最多重试 3次，成功为止
*
* @param map
* 输入参数
*/
private void retry(Map<String, Object> map) {
// 构建重试模板实例
RetryTemplate retryTemplate = new RetryTemplate();
// 设置重试次数
SimpleRetryPolicy policy = new SimpleRetryPolicy(3,
Collections.<Class<? extends Throwable>, Boolean>singletonMap(Exception.class, true));
// 设置重试间隔时间
FixedBackOffPolicy fixedBackOffPolicy = new FixedBackOffPolicy();
fixedBackOffPolicy.setBackOffPeriod(100);
retryTemplate.setRetryPolicy(policy);
retryTemplate.setBackOffPolicy(fixedBackOffPolicy);
// 编写业务处理代码逻辑
final RetryCallback<Object, Exception> retryCallback = new RetryCallback<Object, Exception>() {
public Object doWithRetry(RetryContext context) throws Exception {
System.out.println("第" + (1 + context.getRetryCount()) + "次处理");
try {
bizOp(map);
} catch (Exception e) {
e.printStackTrace();
throw new Exception("捕捉到业务处理异常，需要抛出");// 这个点特别注意，重试的根源通过Exception返回
}
return null;
}
};
// 重试次数执行完依然报错，走如下逻辑
final RecoveryCallback<Object> recoveryCallback = new RecoveryCallback<Object>() {
public Object recover(RetryContext context) throws Exception {
failed(map);
return null;
}
};
try {
// 由retryTemplate 执行execute方法开始逻辑执行
retryTemplate.execute(retryCallback, recoveryCallback);
} catch (Exception e) {
e.printStackTrace();
}
}
}
|