RocketMQ事务消息代码样例

时间 2019-11-16

原文原文链接

本篇主要是分布式工程中，有些跨数据库操做的使用样例，通常可用在分布式事务上。java

MQ的做用，固然有扛洪峰，消息堆集，异步处理的做用。spring

第一步:添加POM的依赖，版本固然由你本身选择数据库

<dependency>
   <groupId>com.alibaba.rocketmq</groupId>
   <artifactId>rocketmq-client</artifactId>
   <version>3.2.6</version>
</dependency>

第二步，消息生产者。json

package com.xxx.consumer.mq;

import com.alibaba.rocketmq.client.QueryResult;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.*;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.common.message.MessageExt;
import org.springframework.stereotype.Component;

import java.util.Map;

@Component
public class MQProducer {
   
   private final String GROUP_NAME = "xxxx";
   private final String NAMESRV_ADDR = "xxx.xxx.xxx.xxx:9876";
   private TransactionMQProducer producer;
   
   public MQProducer() {
      
      this.producer = new TransactionMQProducer(GROUP_NAME);
      this.producer.setNamesrvAddr(NAMESRV_ADDR);    //nameserver服务
      this.producer.setCheckThreadPoolMinSize(5);    // 事务回查最小并发数
      this.producer.setCheckThreadPoolMaxSize(20);   // 事务回查最大并发数
      this.producer.setCheckRequestHoldMax(2000);    // 队列数
      //服务器回调Producer，检查本地事务分支成功仍是失败
      this.producer.setTransactionCheckListener(new TransactionCheckListener() {
         public LocalTransactionState checkLocalTransactionState(MessageExt msg) {
            System.out.println("state -- "+ new String(msg.getBody()));
            return LocalTransactionState.COMMIT_MESSAGE;
         }
      });
      try {
         this.producer.start();
      } catch (MQClientException e) {
         e.printStackTrace();
      }  
   }
   
   public QueryResult queryMessage(String topic, String key, int maxNum, long begin, long end) throws Exception {
      return this.producer.queryMessage(topic, key, maxNum, begin, end);
   }
   
   public LocalTransactionState check(MessageExt me){
      LocalTransactionState ls = this.producer.getTransactionCheckListener().checkLocalTransactionState(me);
      return ls;
   }
   
   public void sendTransactionMessage(Message message, LocalTransactionExecuter localTransactionExecuter, Map<String, Object> transactionMapArgs) throws Exception {
      TransactionSendResult tsr = this.producer.sendMessageInTransaction(message, localTransactionExecuter, transactionMapArgs);
      System.out.println("send返回内容：" + tsr.toString());
   }
   
   public void shutdown(){
      Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
         public void run() {
            producer.shutdown();
         }
      }));
      System.exit(0);
   }


}

组名和地址固然根据你的实际状况写。springboot

第三步，写一个你要执行的方法，好比你的本项目的一次数据库执行，或者其余业务代码。我这里要执行的是保存我的信息。服务器

personInfoService.savePersonalInfo(userid, workClass,workCity);

全代码以下。并发

package com.xxx.consumer.mq;

import com.alibaba.dubbo.config.annotation.Reference;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.rocketmq.client.producer.LocalTransactionExecuter;
import com.alibaba.rocketmq.client.producer.LocalTransactionState;
import com.alibaba.rocketmq.common.message.Message;
import com.wmq.stub.PersonInfoService;
import org.springframework.stereotype.Component;

import java.util.Map;


/**
 * 执行本地事务，由客户端回调
 */

//@Scope("prototype")
@Component("transactionExecuterImpl")
public class TransactionExecuterImpl implements LocalTransactionExecuter {

   @Reference(version = "1.0.0")
   private PersonInfoService personInfoService;
   
   public LocalTransactionState executeLocalTransactionBranch(Message msg, Object arg) {
      try {
         //Message Body
         JSONObject messageBody = FastJsonConvert.convertJSONToObject(new String(msg.getBody(), "utf-8"), JSONObject.class);
         //Transaction MapArgs
         Map<String, Object> mapArgs = (Map<String, Object>) arg;
         
         // --------------------IN PUT---------------------- //
         System.out.println("message body = " + messageBody);
         System.out.println("message mapArgs = " + mapArgs);
         System.out.println("message tag = " + msg.getTags());
         // --------------------IN PUT---------------------- //
         long userid = messageBody.getLong("userid");
         String face = messageBody.getString("face"); //头像
         String trueName = messageBody.getString("truename"); //姓名
         int gender = messageBody.getInteger("gender"); //性别
         int workClass = messageBody.getInteger("workClass"); //指望职位ID
         int workCity = messageBody.getInteger("workCity");
         personInfoService.savePersonalInfo(userid, workClass,workCity);
         //成功通知MQ消息变动 该消息变为：<确认发送>
         
         return LocalTransactionState.COMMIT_MESSAGE;
         
         //return LocalTransactionState.UNKNOW;
         
      } catch (Exception e) {
         e.printStackTrace();
         //失败则不通知MQ 该消息一直处于：<暂缓发送>
         return LocalTransactionState.ROLLBACK_MESSAGE;
         
      }
      
   }
}

注：由于我这里用的是springboot的dubbo框架，因此框架

@Reference(version = "1.0.0")dom

private PersonInfoService personInfoService;是@Reference而不是@AutoWired异步

可改为大家本身的注入。事务消息的本质在这段代码中能够很清楚，发送一条消息出去，而后判断该事务是否执行成功，若成功，通知消息能够发送给消费者，不然该消息暂缓发送。

第四步，在交互代码里面调用

好比某一个controller或者其余地方(。。。。。。。为大家本身的获取数据来源的代码)

try {
    long userid = 。。。。。。。
    String face = 。。。。。。。
    String trueName = 。。。。。。。。
    int gender = 。。。。。。。。
    int workClass = 。。。。。。。。
    int workCity = 。。。。。。。。。
    //构造消息数据
    Message message = new Message();
    //主题
    message.setTopic("user");
    //子标签
    message.setTags("tag");
    //key
    String uuid = UUID.randomUUID().toString();
    message.setKeys(uuid);
    JSONObject body = new JSONObject();
    body.put("userid", userid);
    body.put("face", face);
    body.put("truename", trueName);
    body.put("gender", gender);
    body.put("workClass", workClass);
    body.put("workCity", workCity);
    message.setBody(FastJsonConvert.convertObjectToJSON(body).getBytes());

    //添加参数
    Map<String, Object> transactionMapArgs = new HashMap<String, Object>();
    this.mQProducer.sendTransactionMessage(message, this.transactionExecuterImpl, transactionMapArgs);

} catch (Exception e) {
    e.printStackTrace();
}

这里主要是在消息体中获取参数以及发送消息，这里的消息要等待事务执行成功才能被消费者得到。比较重要的地方就是消息的主题 message.setTopic("user");注意，生产者和消费者的主题必须相同，不然消费者是拿不到消息的，至于主题是什么能够本身定义。

第五步，消费者

消费者通常是写在分布式的另一个工程里面的，并且是不一样的数据库，这一点比较重要，由于同一工程用MQ的意义不大(泄洪除外)。

package com.xxx.xxx.mq;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.rocketmq.client.QueryResult;
import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.xxx.model.User;
import com.xxx.stub.authentication.UpdateUserService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
public class MQConsumer {
   
   private final String GROUP_NAME = "xxxxxxxx";
   private final String NAMESRV_ADDR = "xxx.xxx.xxx.xxx:9876";
   private DefaultMQPushConsumer consumer;
   
   @Autowired
   private UpdateUserService updateUserService;
   
   
   public MQConsumer() {
      try {
         this.consumer = new DefaultMQPushConsumer(GROUP_NAME);
         this.consumer.setNamesrvAddr(NAMESRV_ADDR);
         this.consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
         this.consumer.subscribe("user", "*");
         this.consumer.registerMessageListener(new Listener());
         this.consumer.start();
         System.out.println("consumer start");
      } catch (Exception e) {
         e.printStackTrace();
      }
   }

   public QueryResult queryMessage(String topic, String key, int maxNum, long begin, long end) throws Exception {
      long current = System.currentTimeMillis();
      return this.consumer.queryMessage(topic, key, maxNum, begin, end);
   }
   
   class Listener implements MessageListenerConcurrently {
      public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
         MessageExt msg = msgs.get(0);
         try {
            String topic = msg.getTopic();
            //Message Body
            JSONObject messageBody = FastJsonConvert.convertJSONToObject(new String(msg.getBody(), "utf-8"), JSONObject.class);
            String tags = msg.getTags();
            String keys = msg.getKeys();
            
            System.out.println("服务收到消息, keys : " + keys + ", body : " + new String(msg.getBody(), "utf-8"));
            long userid = messageBody.getLong("userid");
            String face = messageBody.getString("face"); //头像
            String trueName = messageBody.getString("truename"); //姓名
            int gender = messageBody.getInteger("gender"); //性别
            int workClass = messageBody.getInteger("workClass"); //指望职位ID
            int workCity = messageBody.getInteger("workCity");
            User user = new User();
            user.setUserId(userid);
            user.setFace(face);
            user.setTruename(trueName);
            user.setGender(gender);
            user.setStep(4);
            updateUserService.updateUserInfo(user);
            
            
            
         } catch (Exception e) {    
            e.printStackTrace();
            //重试次数为3状况 
            if(msg.getReconsumeTimes() == 3){
               return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
               //记录日志
            }
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
         }        
         return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
      }
   }

}

Listener是一个侦听类，专门负责侦听broker的消息是否到达，而且取回this.consumer.subscribe("user", "*");这个主题的消息。取得消息后执行业务代码updateUserService.updateUserInfo(user);

通常来讲，请严格测试该业务代码，由于rocketmq若是这段业务代码失败，消息是会从新发送给消费者，从新执行这段代码直到成功的，以此到达事务的最终一致性，因此你的这段代码决不能自己就是有Bug的，固然咱们通常会处理3次，并不让他不停的处理，若是仍是失败则记录日志，咱们须要查看日志来解决。