SpringBoot整合rocketmq事务消息

package com.test.rocketmq.mq; import com.alibaba.fastjson.JSONObject; import com.test.rocketmq.dao.pojo.Account; import com.test.rocketmq.dao.pojo.Journal; import com.test.rocketmq.service.AccountService; import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext; import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently; import org.apache.rocketmq.common.message.MessageExt; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import org.springframework.transaction.annotation.Transactional; import java.util.Date; import java.util.List; /** * @author: WangSaiChao * @date: 2018/2/12 * @description: 消息消费者 */ @Component public class MessageListenerImpl implements MessageListenerConcurrently { @Autowired private AccountService accountService; @Override @Transactional public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { MessageExt msg=msgs.get(0); try { String topic = msg.getTopic(); //Message Body JSONObject messageBody=JSONObject.parseObject(new String(msg.getBody(),"UTF-8")); String tags = msg.getTags(); String keys = msg.getKeys(); String userId = messageBody.getString("userId"); String orderId = messageBody.getString("orderId"); Double amount = messageBody.getDouble("amount"); //打印 System.out.println(topic+" "+tags+" "+keys+" 用户id为： "+userId+",订单号为： " + orderId + ",消费金额为："+amount); //业务逻辑处理,根据用户id去数据库中 将账户余额更新 Account account = new Account(); Date date = new Date(); account.setAmount(amount); account.setUserId(userId); account.setUpdateTime(date); accountService.update(account); //保存记录到账户流水表 Journal journal = new Journal(); journal.setOrderId(orderId); journal.setUserId(userId); journal.setAmount(amount); journal.setUpdateTime(date); accountService.insert(journal); return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } catch (Exception e) { e.printStackTrace(); //重试次数为3情况 if(msg.getReconsumeTimes() == 3){ /** * 如果连续重发3次还是发布出去 证明该订单可能哪里有错 * 保存错误日志,然后就不再发送消息了,需要人工介入,逻辑我就不写了 */ return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } //如果没超过三次,稍后再次发送 return ConsumeConcurrentlyStatus.RECONSUME_LATER; } } }