学习来源 https://www.bilibili.com/video/BV1L4411y7mn（添加小部分笔记）感谢作者!基本架构
架构#
流程图#
下单流程#
支付流程#
SpringBoot整合RocketMQ#
依赖包#
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
</dependency>生产者#
yaml#
rocketmq:
name-server: 192.168.1.135:9876;192.168.1.138:9876
producer:
group: my-group使用#
@Autowired
private RocketMQTemplate template;
@RequestMapping("rocketmq")
public String rocketmq(){
log.info("我被调用了-rocketmq");
//主题+内容
template.convertAndSend("mytopic-ly","hello1231");
return "hello world"+serverPort;
}消费者#
yaml#
rocketmq:
name-server: 192.168.1.135:9876;192.168.1.138:9876
consumer:
group: my-group2使用#
创建监听器
@RocketMQMessageListener(topic = "mytopic-ly",
consumeMode = ConsumeMode.CONCURRENTLY,consumerGroup = "${rocketmq.producer.group}")
@Slf4j
@Component
public class Consumer implements RocketMQListener<String> {
@Override
public void onMessage(String s) {
log.info("消费了"+s);
}
}下单流程利用MQ进行回退处理，保证数据一致性#
库存回退的消费者，代码如下：
@Slf4j
@Component
@RocketMQMessageListener(topic = "${mq.order.topic}",consumerGroup =
"${mq.order.consumer.group.name}",messageModel = MessageModel.BROADCASTING )
public class CancelMQListener implements RocketMQListener<MessageExt>{
@Value("${mq.order.consumer.group.name}")
private String groupName;
@Autowired
private TradeGoodsMapper goodsMapper;
@Autowired
private TradeMqConsumerLogMapper mqConsumerLogMapper;
@Autowired
private TradeGoodsNumberLogMapper goodsNumberLogMapper;
@Override
public void onMessage(MessageExt messageExt) {
String msgId=null;
String tags=null;
String keys=null;
String body=null;
try {
//1. 解析消息内容
msgId = messageExt.getMsgId();
tags= messageExt.getTags();
keys= messageExt.getKeys();
body= new String(messageExt.getBody(),"UTF-8");
log.info("接受消息成功");
//2. 查询消息消费记录
TradeMqConsumerLogKey primaryKey = new TradeMqConsumerLogKey();
primaryKey.setMsgTag(tags);
primaryKey.setMsgKey(keys);
primaryKey.setGroupName(groupName);
TradeMqConsumerLog mqConsumerLog = mqConsumerLogMapper.selectByPrimaryKey(primaryKey);
if(mqConsumerLog!=null){
//3. 判断如果消费过...
//3.1 获得消息处理状态
Integer status = mqConsumerLog.getConsumerStatus();
//处理过...返回
if(ShopCode.SHOP_MQ_MESSAGE_STATUS_SUCCESS.getCode().intValue()==status.intValue()){
log.info("消息:"+msgId+",已经处理过");
return;
}
//正在处理...返回
if(ShopCode.SHOP_MQ_MESSAGE_STATUS_PROCESSING.getCode().intValue()==status.intValue()){
log.info("消息:"+msgId+",正在处理");
return;
}
//处理失败
if(ShopCode.SHOP_MQ_MESSAGE_STATUS_FAIL.getCode().intValue()==status.intValue()){
//获得消息处理次数
Integer times = mqConsumerLog.getConsumerTimes();
if(times>3){
log.info("消息:"+msgId+",消息处理超过3次,不能再进行处理了");
return;
}
mqConsumerLog.setConsumerStatus(ShopCode.SHOP_MQ_MESSAGE_STATUS_PROCESSING.getCode());
//使用数据库乐观锁更新
TradeMqConsumerLogExample example = new TradeMqConsumerLogExample();
TradeMqConsumerLogExample.Criteria criteria = example.createCriteria();
criteria.andMsgTagEqualTo(mqConsumerLog.getMsgTag());
criteria.andMsgKeyEqualTo(mqConsumerLog.getMsgKey());
criteria.andGroupNameEqualTo(groupName);
criteria.andConsumerTimesEqualTo(mqConsumerLog.getConsumerTimes());
int r = mqConsumerLogMapper.updateByExampleSelective(mqConsumerLog, example);
if(r<=0){
//未修改成功,其他线程并发修改
log.info("并发修改,稍后处理");
}
}
}else{
//4. 判断如果没有消费过...
mqConsumerLog = new TradeMqConsumerLog();
mqConsumerLog.setMsgTag(tags);
mqConsumerLog.setMsgKey(keys);
mqConsumerLog.setGroupName(groupName);
mqConsumerLog.setConsumerStatus(ShopCode.SHOP_MQ_MESSAGE_STATUS_PROCESSING.getCode());
mqConsumerLog.setMsgBody(body);
mqConsumerLog.setMsgId(msgId);
mqConsumerLog.setConsumerTimes(0);
//将消息处理信息添加到数据库
mqConsumerLogMapper.insert(mqConsumerLog);
}
//5. 回退库存
MQEntity mqEntity = JSON.parseObject(body, MQEntity.class);
Long goodsId = mqEntity.getGoodsId();
TradeGoods goods = goodsMapper.selectByPrimaryKey(goodsId);
goods.setGoodsNumber(goods.getGoodsNumber()+mqEntity.getGoodsNum());
goodsMapper.updateByPrimaryKey(goods);
//6. 将消息的处理状态改为成功
mqConsumerLog.setConsumerStatus(ShopCode.SHOP_MQ_MESSAGE_STATUS_SUCCESS.getCode());
mqConsumerLog.setConsumerTimestamp(new Date());
mqConsumerLogMapper.updateByPrimaryKey(mqConsumerLog);
log.info("回退库存成功");
} catch (Exception e) {
e.printStackTrace();
TradeMqConsumerLogKey primaryKey = new TradeMqConsumerLogKey();
primaryKey.setMsgTag(tags);
primaryKey.setMsgKey(keys);
primaryKey.setGroupName(groupName);
TradeMqConsumerLog mqConsumerLog = mqConsumerLogMapper.selectByPrimaryKey(primaryKey);
if(mqConsumerLog==null){
//数据库未有记录
mqConsumerLog = new TradeMqConsumerLog();
mqConsumerLog.setMsgTag(tags);
mqConsumerLog.setMsgKey(keys);
mqConsumerLog.setGroupName(groupName);
mqConsumerLog.setConsumerStatus(ShopCode.SHOP_MQ_MESSAGE_STATUS_FAIL.getCode());
mqConsumerLog.setMsgBody(body);
mqConsumerLog.setMsgId(msgId);
mqConsumerLog.setConsumerTimes(1);
mqConsumerLogMapper.insert(mqConsumerLog);
}else{
mqConsumerLog.setConsumerTimes(mqConsumerLog.getConsumerTimes()+1);
mqConsumerLogMapper.updateByPrimaryKeySelective(mqConsumerLog);
}
}
}
}
