Spring Cloud微服务系统基于Rocketmq可靠消息最终一致性实现分布式事务

转x33g5p2x  于2021-11-01 转载在 Spring  
字(31.4k)|赞(0)|评价(0)|浏览(663)

一.订单使用事务消息，异步调用账户

订单添加事务消息

1.父项目添加spring Rocketmq依赖

  1. <dependency>
  2.             <groupId>org.apache.rocketmq</groupId>
  3.             <artifactId>rocketmq-spring-boot-starter</artifactId>
  4.             <version>2.1.0</version>
  5.         </dependency>

2.yml配置rocket连接和生产者组名

  1. rocketmq:
  2.   name-server: 192.168.64.141:9876
  3.   producer:
  4.     group: order_producer

3.添加新的数据表tx_table,用来存储事务的执行状态

  1. drop database if exists `seata_order`;
  3. CREATE DATABASE `seata_order` charset utf8;
  5. use `seata_order`;
  7. CREATE TABLE `order` (
  8.   `id` bigint(11) NOT NULL,
  9.   `user_id` bigint(11) DEFAULT NULL COMMENT '用户id',
  10.   `product_id` bigint(11) DEFAULT NULL COMMENT '产品id',
  11.   `count` int(11) DEFAULT NULL COMMENT '数量',
  12.   `money` decimal(11,0) DEFAULT NULL COMMENT '金额',
  13.   PRIMARY KEY (`id`)
  14. ) ENGINE=InnoDB AUTO_INCREMENT=7 DEFAULT CHARSET=utf8;
  16. ALTER TABLE `order` ADD COLUMN `status` int(1) DEFAULT NULL COMMENT '订单状态：0：创建中；1：已完结' AFTER `money` ;
  18. -- for AT mode you must to init this sql for you business database. the seata server not need it.
  19. CREATE TABLE IF NOT EXISTS `undo_log`
  20. (
  21.     `branch_id`     BIGINT(20)   NOT NULL COMMENT 'branch transaction id',
  22.     `xid`           VARCHAR(100) NOT NULL COMMENT 'global transaction id',
  23.     `context`       VARCHAR(128) NOT NULL COMMENT 'undo_log context,such as serialization',
  24.     `rollback_info` LONGBLOB     NOT NULL COMMENT 'rollback info',
  25.     `log_status`    INT(11)      NOT NULL COMMENT '0:normal status,1:defense status',
  26.     `log_created`   DATETIME(6)  NOT NULL COMMENT 'create datetime',
  27.     `log_modified`  DATETIME(6)  NOT NULL COMMENT 'modify datetime',
  28.     UNIQUE KEY `ux_undo_log` (`xid`, `branch_id`)
  29. ) ENGINE = InnoDB
  30.   AUTO_INCREMENT = 1
  31.   DEFAULT CHARSET = utf8 COMMENT ='AT transaction mode undo table';
  33. CREATE TABLE IF NOT EXISTS segment
  34. (
  35.     id            BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY COMMENT '自增主键',
  36.     VERSION       BIGINT      DEFAULT 0  NOT NULL COMMENT '版本号',
  37.     business_type VARCHAR(63) DEFAULT '' NOT NULL COMMENT '业务类型，唯一',
  38.     max_id        BIGINT      DEFAULT 0  NOT NULL COMMENT '当前最大id',
  39.     step          INT         DEFAULT 0  NULL COMMENT '步长',
  40.     increment     INT         DEFAULT 1  NOT NULL COMMENT '每次id增量',
  41.     remainder     INT         DEFAULT 0  NOT NULL COMMENT '余数',
  42.     created_at    BIGINT UNSIGNED        NOT NULL COMMENT '创建时间',
  43.     updated_at    BIGINT UNSIGNED        NOT NULL COMMENT '更新时间',
  44.     CONSTRAINT uniq_business_type UNIQUE (business_type)
  45. ) CHARSET = utf8mb4
  46.   ENGINE INNODB COMMENT '号段表';
  48. INSERT INTO segment
  49. (VERSION, business_type, max_id, step, increment, remainder, created_at, updated_at)
  50. VALUES (1, 'order_business', 1000, 1000, 1, 0, NOW(), NOW());
  52. CREATE TABLE tx_table(
  53. 	`xid` char(32) PRIMARY KEY COMMENT '事务id',
  54. 	`status` int COMMENT '0-提交，1-回滚，2-未知',
  55. 	`created_at` BIGINT UNSIGNED NOT NULL COMMENT '创建时间'
  56. );

4.添加实体类TxInfo和TxMapper

TxMapper.xml

  1. <?xml version="1.0" encoding="UTF-8" ?>
  2. <!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
  3. <mapper namespace="cn.tedu.order.mapper.TxMapper" >
  4.     <resultMap id="BaseResultMap" type="cn.tedu.order.tx.TxInfo" >
  5.         <id column="xid" property="xid" jdbcType="CHAR" />
  6.         <result column="created_at" property="created" jdbcType="BIGINT" />
  7.         <result column="status" property="status" jdbcType="INTEGER"/>
  8.     </resultMap>
  10.     <insert id="insert">
  11.         INSERT INTO `tx_table`(`xid`,`created_at`,`status`) VALUES(#{xid},#{created},#{status});
  12.     </insert>
  13.     <select id="exists" resultType="boolean">
  14.         SELECT COUNT(1) FROM tx_table WHERE xid=#{xid};
  15.     </select>
  16.     <select id="selectById" resultMap="BaseResultMap">
  17.         SELECT `xid`,`created_at`,`status` FROM tx_table WHERE xid=#{xid};
  18.     </select>
  19. </mapper>

5.新建AccountMessage，用来封装发给账户的调用信息：userId，money，txId

  1. package cn.tedu.order.entity;
  3. import lombok.AllArgsConstructor;
  4. import lombok.Data;
  5. import lombok.NoArgsConstructor;
  7. import java.math.BigDecimal;
  8. @Data
  9. @NoArgsConstructor
  10. @AllArgsConstructor
  11. public class AccountMessage {
  12.     private Long userId;
  13.     private BigDecimal money;
  14.     private String xid;
  15. }

6.工具类：JsonUtil

  1. package cn.tedu.order.util;
  3. import java.io.File;
  4. import java.io.FileWriter;
  5. import java.io.IOException;
  6. import java.io.InputStream;
  7. import java.io.InputStreamReader;
  8. import java.io.Writer;
  9. import java.math.BigDecimal;
  10. import java.math.BigInteger;
  11. import java.net.URL;
  12. import java.nio.charset.StandardCharsets;
  13. import java.text.SimpleDateFormat;
  14. import java.util.ArrayList;
  15. import java.util.List;
  17. import org.apache.commons.lang3.StringUtils;
  19. import com.fasterxml.jackson.annotation.JsonInclude;
  20. import com.fasterxml.jackson.core.JsonGenerator;
  21. import com.fasterxml.jackson.core.JsonParser;
  22. import com.fasterxml.jackson.core.JsonProcessingException;
  23. import com.fasterxml.jackson.core.type.TypeReference;
  24. import com.fasterxml.jackson.databind.DeserializationFeature;
  25. import com.fasterxml.jackson.databind.JsonNode;
  26. import com.fasterxml.jackson.databind.ObjectMapper;
  27. import com.fasterxml.jackson.databind.SerializationFeature;
  28. import com.fasterxml.jackson.databind.node.ObjectNode;
  29. import com.fasterxml.jackson.datatype.jdk8.Jdk8Module;
  30. import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
  31. import com.fasterxml.jackson.module.paramnames.ParameterNamesModule;
  33. import lombok.extern.slf4j.Slf4j;
  35. @Slf4j
  36. public class JsonUtil {
  37.     private static ObjectMapper mapper;
  38.     private static JsonInclude.Include DEFAULT_PROPERTY_INCLUSION = JsonInclude.Include.NON_DEFAULT;
  39.     private static boolean IS_ENABLE_INDENT_OUTPUT = false;
  40.     private static String CSV_DEFAULT_COLUMN_SEPARATOR = ",";
  41.     static {
  42.         try {
  43.             initMapper();
  44.             configPropertyInclusion();
  45.             configIndentOutput();
  46.             configCommon();
  47.         } catch (Exception e) {
  48.             log.error("jackson config error", e);
  49.         }
  50.     }
  52.     private static void initMapper() {
  53.         mapper = new ObjectMapper();
  54.     }
  56.     private static void configCommon() {
  57.         config(mapper);
  58.     }
  60.     private static void configPropertyInclusion() {
  61.         mapper.setSerializationInclusion(DEFAULT_PROPERTY_INCLUSION);
  62.     }
  64.     private static void configIndentOutput() {
  65.         mapper.configure(SerializationFeature.INDENT_OUTPUT, IS_ENABLE_INDENT_OUTPUT);
  66.     }
  68.     private static void config(ObjectMapper objectMapper) {
  69.         objectMapper.enable(JsonGenerator.Feature.WRITE_BIGDECIMAL_AS_PLAIN);
  70.         objectMapper.enable(DeserializationFeature.ACCEPT_EMPTY_STRING_AS_NULL_OBJECT);
  71.         objectMapper.enable(DeserializationFeature.ACCEPT_SINGLE_VALUE_AS_ARRAY);
  72.         objectMapper.enable(DeserializationFeature.FAIL_ON_READING_DUP_TREE_KEY);
  73.         objectMapper.enable(DeserializationFeature.FAIL_ON_NUMBERS_FOR_ENUMS);
  74.         objectMapper.disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES);
  75.         objectMapper.disable(DeserializationFeature.FAIL_ON_NULL_FOR_PRIMITIVES);
  76.         objectMapper.disable(SerializationFeature.FAIL_ON_EMPTY_BEANS);
  77.         objectMapper.enable(JsonParser.Feature.ALLOW_COMMENTS);
  78.         objectMapper.disable(JsonGenerator.Feature.ESCAPE_NON_ASCII);
  79.         objectMapper.enable(JsonGenerator.Feature.IGNORE_UNKNOWN);
  80.         objectMapper.enable(JsonParser.Feature.ALLOW_UNQUOTED_FIELD_NAMES);
  81.         objectMapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
  82.         objectMapper.setDateFormat(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"));
  83.         objectMapper.enable(JsonParser.Feature.ALLOW_SINGLE_QUOTES);
  84.         objectMapper.registerModule(new ParameterNamesModule());
  85.         objectMapper.registerModule(new Jdk8Module());
  86.         objectMapper.registerModule(new JavaTimeModule());
  87.     }
  88.     public static void setSerializationInclusion(JsonInclude.Include inclusion) {
  89.         DEFAULT_PROPERTY_INCLUSION = inclusion;
  90.         configPropertyInclusion();
  91.     }
  93.     public static void setIndentOutput(boolean isEnable) {
  94.         IS_ENABLE_INDENT_OUTPUT = isEnable;
  95.         configIndentOutput();
  96.     }
  98.     public static <V> V from(URL url, Class<V> c) {
  99.         try {
  100.             return mapper.readValue(url, c);
  101.         } catch (IOException e) {
  102.             log.error("jackson from error, url: {}, type: {}", url.getPath(), c, e);
  103.             return null;
  104.         }
  105.     }
  107.     public static <V> V from(InputStream inputStream, Class<V> c) {
  108.         try {
  109.             return mapper.readValue(inputStream, c);
  110.         } catch (IOException e) {
  111.             log.error("jackson from error, type: {}", c, e);
  112.             return null;
  113.         }
  114.     }
  116.     public static <V> V from(File file, Class<V> c) {
  117.         try {
  118.             return mapper.readValue(file, c);
  119.         } catch (IOException e) {
  120.             log.error("jackson from error, file path: {}, type: {}", file.getPath(), c, e);
  121.             return null;
  122.         }
  123.     }
  125.     public static <V> V from(Object jsonObj, Class<V> c) {
  126.         try {
  127.             return mapper.readValue(jsonObj.toString(), c);
  128.         } catch (IOException e) {
  129.             log.error("jackson from error, json: {}, type: {}", jsonObj.toString(), c, e);
  130.             return null;
  131.         }
  132.     }
  134.     public static <V> V from(String json, Class<V> c) {
  135.         try {
  136.             return mapper.readValue(json, c);
  137.         } catch (IOException e) {
  138.             log.error("jackson from error, json: {}, type: {}", json, c, e);
  139.             return null;
  140.         }
  141.     }
  143.     public static <V> V from(URL url, TypeReference<V> type) {
  144.         try {
  145.             return mapper.readValue(url, type);
  146.         } catch (IOException e) {
  147.             log.error("jackson from error, url: {}, type: {}", url.getPath(), type, e);
  148.             return null;
  149.         }
  150.     }
  152.     public static <V> V from(InputStream inputStream, TypeReference<V> type) {
  153.         try {
  154.             return mapper.readValue(inputStream, type);
  155.         } catch (IOException e) {
  156.             log.error("jackson from error, type: {}", type, e);
  157.             return null;
  158.         }
  159.     }
  161.     public static <V> V from(File file, TypeReference<V> type) {
  162.         try {
  163.             return mapper.readValue(file, type);
  164.         } catch (IOException e) {
  165.             log.error("jackson from error, file path: {}, type: {}", file.getPath(), type, e);
  166.             return null;
  167.         }
  168.     }
  170.     public static <V> V from(Object jsonObj, TypeReference<V> type) {
  171.         try {
  172.             return mapper.readValue(jsonObj.toString(), type);
  173.         } catch (IOException e) {
  174.             log.error("jackson from error, json: {}, type: {}", jsonObj.toString(), type, e);
  175.             return null;
  176.         }
  177.     }
  179.     public static <V> V from(String json, TypeReference<V> type) {
  180.         try {
  181.             return mapper.readValue(json, type);
  182.         } catch (IOException e) {
  183.             log.error("jackson from error, json: {}, type: {}", json, type, e);
  184.             return null;
  185.         }
  186.     }
  188.     public static <V> String to(List<V> list) {
  189.         try {
  190.             return mapper.writeValueAsString(list);
  191.         } catch (JsonProcessingException e) {
  192.             log.error("jackson to error, obj: {}", list, e);
  193.             return null;
  194.         }
  195.     }
  197.     public static <V> String to(V v) {
  198.         try {
  199.             return mapper.writeValueAsString(v);
  200.         } catch (JsonProcessingException e) {
  201.             log.error("jackson to error, obj: {}", v, e);
  202.             return null;
  203.         }
  204.     }
  206.     public static <V> void toFile(String path, List<V> list) {
  207.         try (Writer writer = new FileWriter(new File(path), true)) {
  208.             mapper.writer().writeValues(writer).writeAll(list);
  209.             writer.flush();
  210.         } catch (Exception e) {
  211.             log.error("jackson to file error, path: {}, list: {}", path, list, e);
  212.         }
  213.     }
  215.     public static <V> void toFile(String path, V v) {
  216.         try (Writer writer = new FileWriter(new File(path), true)) {
  217.             mapper.writer().writeValues(writer).write(v);
  218.             writer.flush();
  219.         } catch (Exception e) {
  220.             log.error("jackson to file error, path: {}, obj: {}", path, v, e);
  221.         }
  222.     }
  224.     public static String getString(String json, String key) {
  225.         if (StringUtils.isEmpty(json)) {
  226.             return null;
  227.         }
  228.         try {
  229.             JsonNode node = mapper.readTree(json);
  230.             if (null != node) {
  231.                 return node.get(key).asText();
  232.             } else {
  233.                 return null;
  234.             }
  235.         } catch (IOException e) {
  236.             log.error("jackson get string error, json: {}, key: {}", json, key, e);
  237.             return null;
  238.         }
  239.     }
  241.     public static Integer getInt(String json, String key) {
  242.         if (StringUtils.isEmpty(json)) {
  243.             return null;
  244.         }
  245.         try {
  246.             JsonNode node = mapper.readTree(json);
  247.             if (null != node) {
  248.                 return node.get(key).intValue();
  249.             } else {
  250.                 return null;
  251.             }
  252.         } catch (IOException e) {
  253.             log.error("jackson get int error, json: {}, key: {}", json, key, e);
  254.             return null;
  255.         }
  256.     }
  258.     public static Long getLong(String json, String key) {
  259.         if (StringUtils.isEmpty(json)) {
  260.             return null;
  261.         }
  262.         try {
  263.             JsonNode node = mapper.readTree(json);
  264.             if (null != node) {
  265.                 return node.get(key).longValue();
  266.             } else {
  267.                 return null;
  268.             }
  269.         } catch (IOException e) {
  270.             log.error("jackson get long error, json: {}, key: {}", json, key, e);
  271.             return null;
  272.         }
  273.     }
  275.     public static Double getDouble(String json, String key) {
  276.         if (StringUtils.isEmpty(json)) {
  277.             return null;
  278.         }
  279.         try {
  280.             JsonNode node = mapper.readTree(json);
  281.             if (null != node) {
  282.                 return node.get(key).doubleValue();
  283.             } else {
  284.                 return null;
  285.             }
  286.         } catch (IOException e) {
  287.             log.error("jackson get double error, json: {}, key: {}", json, key, e);
  288.             return null;
  289.         }
  290.     }
  292.     public static BigInteger getBigInteger(String json, String key) {
  293.         if (StringUtils.isEmpty(json)) {
  294.             return new BigInteger(String.valueOf(0.00));
  295.         }
  296.         try {
  297.             JsonNode node = mapper.readTree(json);
  298.             if (null != node) {
  299.                 return node.get(key).bigIntegerValue();
  300.             } else {
  301.                 return null;
  302.             }
  303.         } catch (IOException e) {
  304.             log.error("jackson get biginteger error, json: {}, key: {}", json, key, e);
  305.             return null;
  306.         }
  307.     }
  309.     public static BigDecimal getBigDecimal(String json, String key) {
  310.         if (StringUtils.isEmpty(json)) {
  311.             return null;
  312.         }
  313.         try {
  314.             JsonNode node = mapper.readTree(json);
  315.             if (null != node) {
  316.                 return node.get(key).decimalValue();
  317.             } else {
  318.                 return null;
  319.             }
  320.         } catch (IOException e) {
  321.             log.error("jackson get bigdecimal error, json: {}, key: {}", json, key, e);
  322.             return null;
  323.         }
  324.     }
  326.     public static boolean getBoolean(String json, String key) {
  327.         if (StringUtils.isEmpty(json)) {
  328.             return false;
  329.         }
  330.         try {
  331.             JsonNode node = mapper.readTree(json);
  332.             if (null != node) {
  333.                 return node.get(key).booleanValue();
  334.             } else {
  335.                 return false;
  336.             }
  337.         } catch (IOException e) {
  338.             log.error("jackson get boolean error, json: {}, key: {}", json, key, e);
  339.             return false;
  340.         }
  341.     }
  343.     public static byte[] getByte(String json, String key) {
  344.         if (StringUtils.isEmpty(json)) {
  345.             return null;
  346.         }
  347.         try {
  348.             JsonNode node = mapper.readTree(json);
  349.             if (null != node) {
  350.                 return node.get(key).binaryValue();
  351.             } else {
  352.                 return null;
  353.             }
  354.         } catch (IOException e) {
  355.             log.error("jackson get byte error, json: {}, key: {}", json, key, e);
  356.             return null;
  357.         }
  358.     }
  360.     public static <T> ArrayList<T> getList(String json, String key) {
  361.         if (StringUtils.isEmpty(json)) {
  362.             return null;
  363.         }
  364.         String string = getString(json, key);
  365.         return from(string, new TypeReference<ArrayList<T>>() {});
  366.     }
  368.     public static <T> String add(String json, String key, T value) {
  369.         try {
  370.             JsonNode node = mapper.readTree(json);
  371.             add(node, key, value);
  372.             return node.toString();
  373.         } catch (IOException e) {
  374.             log.error("jackson add error, json: {}, key: {}, value: {}", json, key, value, e);
  375.             return json;
  376.         }
  377.     }
  379.     private static <T> void add(JsonNode jsonNode, String key, T value) {
  380.         if (value instanceof String) {
  381.             ((ObjectNode) jsonNode).put(key, (String) value);
  382.         } else if (value instanceof Short) {
  383.             ((ObjectNode) jsonNode).put(key, (Short) value);
  384.         } else if (value instanceof Integer) {
  385.             ((ObjectNode) jsonNode).put(key, (Integer) value);
  386.         } else if (value instanceof Long) {
  387.             ((ObjectNode) jsonNode).put(key, (Long) value);
  388.         } else if (value instanceof Float) {
  389.             ((ObjectNode) jsonNode).put(key, (Float) value);
  390.         } else if (value instanceof Double) {
  391.             ((ObjectNode) jsonNode).put(key, (Double) value);
  392.         } else if (value instanceof BigDecimal) {
  393.             ((ObjectNode) jsonNode).put(key, (BigDecimal) value);
  394.         } else if (value instanceof BigInteger) {
  395.             ((ObjectNode) jsonNode).put(key, (BigInteger) value);
  396.         } else if (value instanceof Boolean) {
  397.             ((ObjectNode) jsonNode).put(key, (Boolean) value);
  398.         } else if (value instanceof byte[]) {
  399.             ((ObjectNode) jsonNode).put(key, (byte[]) value);
  400.         } else {
  401.             ((ObjectNode) jsonNode).put(key, to(value));
  402.         }
  403.     }
  405.     public static String remove(String json, String key) {
  406.         try {
  407.             JsonNode node = mapper.readTree(json);
  408.             ((ObjectNode) node).remove(key);
  409.             return node.toString();
  410.         } catch (IOException e) {
  411.             log.error("jackson remove error, json: {}, key: {}", json, key, e);
  412.             return json;
  413.         }
  414.     }
  416.     public static <T> String update(String json, String key, T value) {
  417.         try {
  418.             JsonNode node = mapper.readTree(json);
  419.             ((ObjectNode) node).remove(key);
  420.             add(node, key, value);
  421.             return node.toString();
  422.         } catch (IOException e) {
  423.             log.error("jackson update error, json: {}, key: {}, value: {}", json, key, value, e);
  424.             return json;
  425.         }
  426.     }
  428.     public static String format(String json) {
  429.         try {
  430.             JsonNode node = mapper.readTree(json);
  431.             return mapper.writerWithDefaultPrettyPrinter().writeValueAsString(node);
  432.         } catch (IOException e) {
  433.             log.error("jackson format json error, json: {}", json, e);
  434.             return json;
  435.         }
  436.     }
  438.     public static boolean isJson(String json) {
  439.         try {
  440.             mapper.readTree(json);
  441.             return true;
  442.         } catch (Exception e) {
  443.             log.error("jackson check json error, json: {}", json, e);
  444.             return false;
  445.         }
  446.     }
  448.     private static InputStream getResourceStream(String name) {
  449.         return JsonUtil.class.getClassLoader().getResourceAsStream(name);
  450.     }
  452.     private static InputStreamReader getResourceReader(InputStream inputStream) {
  453.         if (null == inputStream) {
  454.             return null;
  455.         }
  456.         return new InputStreamReader(inputStream, StandardCharsets.UTF_8);
  457.     }
  458. }

7.OrderServiceImpl发送事务消息

  1. package cn.tedu.order.service;
  3. import cn.tedu.order.entity.AccountMessage;
  4. import cn.tedu.order.entity.Order;
  5. import cn.tedu.order.entity.TxInfo;
  7. import cn.tedu.order.fegin.AccountClient;
  8. import cn.tedu.order.fegin.EasyIdClient;
  9. import cn.tedu.order.fegin.StorageClient;
  10. import cn.tedu.order.mapper.OrderMapper;
  11. import cn.tedu.order.mapper.TxMapper;
  12. import cn.tedu.order.util.JsonUtil;
  13. import lombok.extern.slf4j.Slf4j;
  14. import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
  15. import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
  16. import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
  17. import org.apache.rocketmq.spring.core.RocketMQTemplate;
  18. import org.springframework.beans.factory.annotation.Autowired;
  19. import org.springframework.messaging.Message;
  20. import org.springframework.messaging.support.MessageBuilder;
  21. import org.springframework.stereotype.Service;
  22. import org.springframework.transaction.annotation.Transactional;
  24. import java.util.Random;
  25. import java.util.UUID;
  27. @Slf4j
  28. @Service
  29. @RocketMQTransactionListener
  30. public class OrderServiceImpl
  31.         implements OrderService, RocketMQLocalTransactionListener {
  33.     @Autowired
  34.     private OrderMapper orderMapper;
  36.     @Autowired
  37.     private EasyIdClient easyIdClient;
  38.     @Autowired
  39.     private AccountClient accountClient;
  40.     @Autowired
  41.     private StorageClient storageClient;
  42.     @Autowired
  43.     private RocketMQTemplate t;
  44.     @Autowired
  45.     private TxMapper txMapper;
  47.     // 业务方法，不直接完成业务，而是发送事务消息
  48.     // 通过发送事务消息，会触发监听器执行业务
  49.     @Override
  50.     public void create(Order order) {
  51.         // 准备消息数据
  52.         String xid = UUID.randomUUID().toString().replace("-", "");
  53.         AccountMessage am = new AccountMessage(order.getUserId(),order.getMoney(),xid);
  54.         String json = JsonUtil.to(am);
  55.         // 把 json 字符串，封装到spring 的通用 Message 对象
  56.         Message<String> msg = MessageBuilder.withPayload(json).build();
  58.         // 发送事务消息
  59.         // 如果有标签： orderTopic:TagA
  60.         // t.sendMessageInTransaction("orderTopic", msg, 触发监听器执行业务时需要的业务数据参数);
  61.         t.sendMessageInTransaction("orderTopic", msg, order);
  62.     }
  64.     public void doCreate(Order order) {
  65.         // 远程调用发号器，生成订单id
  66.         String s = easyIdClient.nextId("order_business");
  67.         Long id = Long.valueOf(s);
  69.         order.setId(id);
  70.         orderMapper.create(order);
  71.     }
  73.     // 执行本地事务
  74.     @Transactional
  75.     @Override
  76.     public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
  77.         RocketMQLocalTransactionState state; // 用来返回状态
  78.         Integer status; // 用来在数据库中保存状态
  80.         try {
  81.             doCreate((Order) o);
  82.             state = RocketMQLocalTransactionState.COMMIT;
  83.             status = 0;
  84.         } catch (Exception e) {
  85.             log.error("创建订单失败", e);
  86.             state = RocketMQLocalTransactionState.ROLLBACK;
  87.             status = 1;
  88.         }
  90.         // message - {userId:6, money:20, xid:j5hg345ytg}
  91.         String json = new String((byte[]) message.getPayload());
  92.         String xid = JsonUtil.getString(json, "xid");
  94.         txMapper.insert(new TxInfo(xid, status, System.currentTimeMillis()));
  95.         return state;
  96.     }
  98.     // 处理事务状态回查
  99.     @Override
  100.     public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
  101.         String json = new String((byte[]) message.getPayload());
  102.         String xid = JsonUtil.getString(json, "xid");
  104.         TxInfo txInfo = txMapper.selectById(xid);
  105.         if (txInfo == null) {
  106.             return RocketMQLocalTransactionState.UNKNOWN;
  107.         }
  108.         switch (txInfo.getStatus()) {
  109.             case 0: return RocketMQLocalTransactionState.COMMIT;
  110.             case 1: return RocketMQLocalTransactionState.ROLLBACK;
  111.             default: return RocketMQLocalTransactionState.UNKNOWN;
  112.         }
  113.     }
  114. }

8.实现事务监听器

TxListener 事务监听器
发送事务消息后会触发事务监听器执行。

事务监听器有两个方法：

  • executeLocalTransaction()： 执行本地事务
  • checkLocalTransaction()： 负责响应Rocketmq服务器的事务回查操作

9.测试

按顺序启动项目：

1.Eureka
2.Easy Id Generator
3.Order
调用保存订单，地址：

观察控制台日志：

订单表：

事务表：

二.账户接收消息，扣减账户

1.依赖

2.yml配置name server

  1. rocketmq:
  2.   name-server: 192.168.64.141:9876

3.AccountMessage

  1. package cn.tedu.account.entity;
  3. import lombok.AllArgsConstructor;
  4. import lombok.Data;
  5. import lombok.NoArgsConstructor;
  7. import java.math.BigDecimal;
  9. @Data
  10. @NoArgsConstructor
  11. @AllArgsConstructor
  12. public class AccountMessage {
  13.     private Long userId;
  14.     private BigDecimal money;
  15.     private String xid;
  16. }

4.jsonUtil

  1. package cn.tedu.account.util;
  3. import com.fasterxml.jackson.annotation.JsonInclude;
  4. import com.fasterxml.jackson.core.JsonGenerator;
  5. import com.fasterxml.jackson.core.JsonParser;
  6. import com.fasterxml.jackson.core.JsonProcessingException;
  7. import com.fasterxml.jackson.core.type.TypeReference;
  8. import com.fasterxml.jackson.databind.DeserializationFeature;
  9. import com.fasterxml.jackson.databind.JsonNode;
  10. import com.fasterxml.jackson.databind.ObjectMapper;
  11. import com.fasterxml.jackson.databind.SerializationFeature;
  12. import com.fasterxml.jackson.databind.node.ObjectNode;
  13. import com.fasterxml.jackson.datatype.jdk8.Jdk8Module;
  14. import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
  15. import com.fasterxml.jackson.module.paramnames.ParameterNamesModule;
  16. import lombok.extern.slf4j.Slf4j;
  17. import org.apache.commons.lang3.StringUtils;
  19. import java.io.*;
  20. import java.math.BigDecimal;
  21. import java.math.BigInteger;
  22. import java.net.URL;
  23. import java.nio.charset.StandardCharsets;
  24. import java.text.SimpleDateFormat;
  25. import java.util.ArrayList;
  26. import java.util.List;
  28. @Slf4j
  29. public class JsonUtil {
  30.     private static ObjectMapper mapper;
  31.     private static JsonInclude.Include DEFAULT_PROPERTY_INCLUSION = JsonInclude.Include.NON_DEFAULT;
  32.     private static boolean IS_ENABLE_INDENT_OUTPUT = false;
  33.     private static String CSV_DEFAULT_COLUMN_SEPARATOR = ",";
  34.     static {
  35.         try {
  36.             initMapper();
  37.             configPropertyInclusion();
  38.             configIndentOutput();
  39.             configCommon();
  40.         } catch (Exception e) {
  41.             log.error("jackson config error", e);
  42.         }
  43.     }
  45.     private static void initMapper() {
  46.         mapper = new ObjectMapper();
  47.     }
  49.     private static void configCommon() {
  50.         config(mapper);
  51.     }
  53.     private static void configPropertyInclusion() {
  54.         mapper.setSerializationInclusion(DEFAULT_PROPERTY_INCLUSION);
  55.     }
  57.     private static void configIndentOutput() {
  58.         mapper.configure(SerializationFeature.INDENT_OUTPUT, IS_ENABLE_INDENT_OUTPUT);
  59.     }
  61.     private static void config(ObjectMapper objectMapper) {
  62.         objectMapper.enable(JsonGenerator.Feature.WRITE_BIGDECIMAL_AS_PLAIN);
  63.         objectMapper.enable(DeserializationFeature.ACCEPT_EMPTY_STRING_AS_NULL_OBJECT);
  64.         objectMapper.enable(DeserializationFeature.ACCEPT_SINGLE_VALUE_AS_ARRAY);
  65.         objectMapper.enable(DeserializationFeature.FAIL_ON_READING_DUP_TREE_KEY);
  66.         objectMapper.enable(DeserializationFeature.FAIL_ON_NUMBERS_FOR_ENUMS);
  67.         objectMapper.disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES);
  68.         objectMapper.disable(DeserializationFeature.FAIL_ON_NULL_FOR_PRIMITIVES);
  69.         objectMapper.disable(SerializationFeature.FAIL_ON_EMPTY_BEANS);
  70.         objectMapper.enable(JsonParser.Feature.ALLOW_COMMENTS);
  71.         objectMapper.disable(JsonGenerator.Feature.ESCAPE_NON_ASCII);
  72.         objectMapper.enable(JsonGenerator.Feature.IGNORE_UNKNOWN);
  73.         objectMapper.enable(JsonParser.Feature.ALLOW_UNQUOTED_FIELD_NAMES);
  74.         objectMapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
  75.         objectMapper.setDateFormat(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"));
  76.         objectMapper.enable(JsonParser.Feature.ALLOW_SINGLE_QUOTES);
  77.         objectMapper.registerModule(new ParameterNamesModule());
  78.         objectMapper.registerModule(new Jdk8Module());
  79.         objectMapper.registerModule(new JavaTimeModule());
  80.     }
  81.     public static void setSerializationInclusion(JsonInclude.Include inclusion) {
  82.         DEFAULT_PROPERTY_INCLUSION = inclusion;
  83.         configPropertyInclusion();
  84.     }
  86.     public static void setIndentOutput(boolean isEnable) {
  87.         IS_ENABLE_INDENT_OUTPUT = isEnable;
  88.         configIndentOutput();
  89.     }
  91.     public static <V> V from(URL url, Class<V> c) {
  92.         try {
  93.             return mapper.readValue(url, c);
  94.         } catch (IOException e) {
  95.             log.error("jackson from error, url: {}, type: {}", url.getPath(), c, e);
  96.             return null;
  97.         }
  98.     }
  100.     public static <V> V from(InputStream inputStream, Class<V> c) {
  101.         try {
  102.             return mapper.readValue(inputStream, c);
  103.         } catch (IOException e) {
  104.             log.error("jackson from error, type: {}", c, e);
  105.             return null;
  106.         }
  107.     }
  109.     public static <V> V from(File file, Class<V> c) {
  110.         try {
  111.             return mapper.readValue(file, c);
  112.         } catch (IOException e) {
  113.             log.error("jackson from error, file path: {}, type: {}", file.getPath(), c, e);
  114.             return null;
  115.         }
  116.     }
  118.     public static <V> V from(Object jsonObj, Class<V> c) {
  119.         try {
  120.             return mapper.readValue(jsonObj.toString(), c);
  121.         } catch (IOException e) {
  122.             log.error("jackson from error, json: {}, type: {}", jsonObj.toString(), c, e);
  123.             return null;
  124.         }
  125.     }
  127.     public static <V> V from(String json, Class<V> c) {
  128.         try {
  129.             return mapper.readValue(json, c);
  130.         } catch (IOException e) {
  131.             log.error("jackson from error, json: {}, type: {}", json, c, e);
  132.             return null;
  133.         }
  134.     }
  136.     public static <V> V from(URL url, TypeReference<V> type) {
  137.         try {
  138.             return mapper.readValue(url, type);
  139.         } catch (IOException e) {
  140.             log.error("jackson from error, url: {}, type: {}", url.getPath(), type, e);
  141.             return null;
  142.         }
  143.     }
  145.     public static <V> V from(InputStream inputStream, TypeReference<V> type) {
  146.         try {
  147.             return mapper.readValue(inputStream, type);
  148.         } catch (IOException e) {
  149.             log.error("jackson from error, type: {}", type, e);
  150.             return null;
  151.         }
  152.     }
  154.     public static <V> V from(File file, TypeReference<V> type) {
  155.         try {
  156.             return mapper.readValue(file, type);
  157.         } catch (IOException e) {
  158.             log.error("jackson from error, file path: {}, type: {}", file.getPath(), type, e);
  159.             return null;
  160.         }
  161.     }
  163.     public static <V> V from(Object jsonObj, TypeReference<V> type) {
  164.         try {
  165.             return mapper.readValue(jsonObj.toString(), type);
  166.         } catch (IOException e) {
  167.             log.error("jackson from error, json: {}, type: {}", jsonObj.toString(), type, e);
  168.             return null;
  169.         }
  170.     }
  172.     public static <V> V from(String json, TypeReference<V> type) {
  173.         try {
  174.             return mapper.readValue(json, type);
  175.         } catch (IOException e) {
  176.             log.error("jackson from error, json: {}, type: {}", json, type, e);
  177.             return null;
  178.         }
  179.     }
  181.     public static <V> String to(List<V> list) {
  182.         try {
  183.             return mapper.writeValueAsString(list);
  184.         } catch (JsonProcessingException e) {
  185.             log.error("jackson to error, obj: {}", list, e);
  186.             return null;
  187.         }
  188.     }
  190.     public static <V> String to(V v) {
  191.         try {
  192.             return mapper.writeValueAsString(v);
  193.         } catch (JsonProcessingException e) {
  194.             log.error("jackson to error, obj: {}", v, e);
  195.             return null;
  196.         }
  197.     }
  199.     public static <V> void toFile(String path, List<V> list) {
  200.         try (Writer writer = new FileWriter(new File(path), true)) {
  201.             mapper.writer().writeValues(writer).writeAll(list);
  202.             writer.flush();
  203.         } catch (Exception e) {
  204.             log.error("jackson to file error, path: {}, list: {}", path, list, e);
  205.         }
  206.     }
  208.     public static <V> void toFile(String path, V v) {
  209.         try (Writer writer = new FileWriter(new File(path), true)) {
  210.             mapper.writer().writeValues(writer).write(v);
  211.             writer.flush();
  212.         } catch (Exception e) {
  213.             log.error("jackson to file error, path: {}, obj: {}", path, v, e);
  214.         }
  215.     }
  217.     public static String getString(String json, String key) {
  218.         if (StringUtils.isEmpty(json)) {
  219.             return null;
  220.         }
  221.         try {
  222.             JsonNode node = mapper.readTree(json);
  223.             if (null != node) {
  224.                 return node.get(key).asText();
  225.             } else {
  226.                 return null;
  227.             }
  228.         } catch (IOException e) {
  229.             log.error("jackson get string error, json: {}, key: {}", json, key, e);
  230.             return null;
  231.         }
  232.     }
  234.     public static Integer getInt(String json, String key) {
  235.         if (StringUtils.isEmpty(json)) {
  236.             return null;
  237.         }
  238.         try {
  239.             JsonNode node = mapper.readTree(json);
  240.             if (null != node) {
  241.                 return node.get(key).intValue();
  242.             } else {
  243.                 return null;
  244.             }
  245.         } catch (IOException e) {
  246.             log.error("jackson get int error, json: {}, key: {}", json, key, e);
  247.             return null;
  248.         }
  249.     }
  251.     public static Long getLong(String json, String key) {
  252.         if (StringUtils.isEmpty(json)) {
  253.             return null;
  254.         }
  255.         try {
  256.             JsonNode node = mapper.readTree(json);
  257.             if (null != node) {
  258.                 return node.get(key).longValue();
  259.             } else {
  260.                 return null;
  261.             }
  262.         } catch (IOException e) {
  263.             log.error("jackson get long error, json: {}, key: {}", json, key, e);
  264.             return null;
  265.         }
  266.     }
  268.     public static Double getDouble(String json, String key) {
  269.         if (StringUtils.isEmpty(json)) {
  270.             return null;
  271.         }
  272.         try {
  273.             JsonNode node = mapper.readTree(json);
  274.             if (null != node) {
  275.                 return node.get(key).doubleValue();
  276.             } else {
  277.                 return null;
  278.             }
  279.         } catch (IOException e) {
  280.             log.error("jackson get double error, json: {}, key: {}", json, key, e);
  281.             return null;
  282.         }
  283.     }
  285.     public static BigInteger getBigInteger(String json, String key) {
  286.         if (StringUtils.isEmpty(json)) {
  287.             return new BigInteger(String.valueOf(0.00));
  288.         }
  289.         try {
  290.             JsonNode node = mapper.readTree(json);
  291.             if (null != node) {
  292.                 return node.get(key).bigIntegerValue();
  293.             } else {
  294.                 return null;
  295.             }
  296.         } catch (IOException e) {
  297.             log.error("jackson get biginteger error, json: {}, key: {}", json, key, e);
  298.             return null;
  299.         }
  300.     }
  302.     public static BigDecimal getBigDecimal(String json, String key) {
  303.         if (StringUtils.isEmpty(json)) {
  304.             return null;
  305.         }
  306.         try {
  307.             JsonNode node = mapper.readTree(json);
  308.             if (null != node) {
  309.                 return node.get(key).decimalValue();
  310.             } else {
  311.                 return null;
  312.             }
  313.         } catch (IOException e) {
  314.             log.error("jackson get bigdecimal error, json: {}, key: {}", json, key, e);
  315.             return null;
  316.         }
  317.     }
  319.     public static boolean getBoolean(String json, String key) {
  320.         if (StringUtils.isEmpty(json)) {
  321.             return false;
  322.         }
  323.         try {
  324.             JsonNode node = mapper.readTree(json);
  325.             if (null != node) {
  326.                 return node.get(key).booleanValue();
  327.             } else {
  328.                 return false;
  329.             }
  330.         } catch (IOException e) {
  331.             log.error("jackson get boolean error, json: {}, key: {}", json, key, e);
  332.             return false;
  333.         }
  334.     }
  336.     public static byte[] getByte(String json, String key) {
  337.         if (StringUtils.isEmpty(json)) {
  338.             return null;
  339.         }
  340.         try {
  341.             JsonNode node = mapper.readTree(json);
  342.             if (null != node) {
  343.                 return node.get(key).binaryValue();
  344.             } else {
  345.                 return null;
  346.             }
  347.         } catch (IOException e) {
  348.             log.error("jackson get byte error, json: {}, key: {}", json, key, e);
  349.             return null;
  350.         }
  351.     }
  353.     public static <T> ArrayList<T> getList(String json, String key) {
  354.         if (StringUtils.isEmpty(json)) {
  355.             return null;
  356.         }
  357.         String string = getString(json, key);
  358.         return from(string, new TypeReference<ArrayList<T>>() {});
  359.     }
  361.     public static <T> String add(String json, String key, T value) {
  362.         try {
  363.             JsonNode node = mapper.readTree(json);
  364.             add(node, key, value);
  365.             return node.toString();
  366.         } catch (IOException e) {
  367.             log.error("jackson add error, json: {}, key: {}, value: {}", json, key, value, e);
  368.             return json;
  369.         }
  370.     }
  372.     private static <T> void add(JsonNode jsonNode, String key, T value) {
  373.         if (value instanceof String) {
  374.             ((ObjectNode) jsonNode).put(key, (String) value);
  375.         } else if (value instanceof Short) {
  376.             ((ObjectNode) jsonNode).put(key, (Short) value);
  377.         } else if (value instanceof Integer) {
  378.             ((ObjectNode) jsonNode).put(key, (Integer) value);
  379.         } else if (value instanceof Long) {
  380.             ((ObjectNode) jsonNode).put(key, (Long) value);
  381.         } else if (value instanceof Float) {
  382.             ((ObjectNode) jsonNode).put(key, (Float) value);
  383.         } else if (value instanceof Double) {
  384.             ((ObjectNode) jsonNode).put(key, (Double) value);
  385.         } else if (value instanceof BigDecimal) {
  386.             ((ObjectNode) jsonNode).put(key, (BigDecimal) value);
  387.         } else if (value instanceof BigInteger) {
  388.             ((ObjectNode) jsonNode).put(key, (BigInteger) value);
  389.         } else if (value instanceof Boolean) {
  390.             ((ObjectNode) jsonNode).put(key, (Boolean) value);
  391.         } else if (value instanceof byte[]) {
  392.             ((ObjectNode) jsonNode).put(key, (byte[]) value);
  393.         } else {
  394.             ((ObjectNode) jsonNode).put(key, to(value));
  395.         }
  396.     }
  398.     public static String remove(String json, String key) {
  399.         try {
  400.             JsonNode node = mapper.readTree(json);
  401.             ((ObjectNode) node).remove(key);
  402.             return node.toString();
  403.         } catch (IOException e) {
  404.             log.error("jackson remove error, json: {}, key: {}", json, key, e);
  405.             return json;
  406.         }
  407.     }
  409.     public static <T> String update(String json, String key, T value) {
  410.         try {
  411.             JsonNode node = mapper.readTree(json);
  412.             ((ObjectNode) node).remove(key);
  413.             add(node, key, value);
  414.             return node.toString();
  415.         } catch (IOException e) {
  416.             log.error("jackson update error, json: {}, key: {}, value: {}", json, key, value, e);
  417.             return json;
  418.         }
  419.     }
  421.     public static String format(String json) {
  422.         try {
  423.             JsonNode node = mapper.readTree(json);
  424.             return mapper.writerWithDefaultPrettyPrinter().writeValueAsString(node);
  425.         } catch (IOException e) {
  426.             log.error("jackson format json error, json: {}", json, e);
  427.             return json;
  428.         }
  429.     }
  431.     public static boolean isJson(String json) {
  432.         try {
  433.             mapper.readTree(json);
  434.             return true;
  435.         } catch (Exception e) {
  436.             log.error("jackson check json error, json: {}", json, e);
  437.             return false;
  438.         }
  439.     }
  441.     private static InputStream getResourceStream(String name) {
  442.         return JsonUtil.class.getClassLoader().getResourceAsStream(name);
  443.     }
  445.     private static InputStreamReader getResourceReader(InputStream inputStream) {
  446.         if (null == inputStream) {
  447.             return null;
  448.         }
  449.         return new InputStreamReader(inputStream, StandardCharsets.UTF_8);
  450.     }
  451. }

5. 新建消费者类： AccountConsumer，实现消费者接口

  1. package cn.tedu.account.tx;
  3. import cn.tedu.account.entity.AccountMessage;
  4. import cn.tedu.account.mapper.AccountMapper;
  5. import cn.tedu.account.service.AccountService;
  6. import cn.tedu.account.util.JsonUtil;
  7. import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
  8. import org.apache.rocketmq.spring.core.RocketMQListener;
  9. import org.springframework.beans.factory.annotation.Autowired;
  10. import org.springframework.stereotype.Component;
  12. @Component
  13. @RocketMQMessageListener(topic = "orderTopic",
  14.                         consumerGroup = "account-consumer")
  16. public class AccountConsumer implements RocketMQListener<String> {
  18.     @Autowired
  19.     private AccountService accountService;
  21.     @Override
  22.     public void onMessage(String json) {
  23.         //json--->AccountMessage
  24.         AccountMessage am = JsonUtil.from(json, AccountMessage.class);
  26.         accountService.decrease(am.getUserId(), am.getMoney());
  28.     }
  29. }

6. 通过注解配置接收消息

  1. package cn.tedu.account.service;
  3. import cn.tedu.account.mapper.AccountMapper;
  4.  import org.springframework.beans.factory.annotation.Autowired;
  5.  import org.springframework.stereotype.Service;
  7. import java.math.BigDecimal;
  9. @Service
  10.  public class AccountServiceImpl implements AccountService{
  11.  @Autowired
  12.  private AccountMapper accountMapper;
  14. @Override
  15. public void decrease(Long userId, BigDecimal money) {
  16.     accountMapper.decrease(userId, money);
  17. }

}

7. 扣减账户

TxConsumer 实现消息监听，收到消息后完成扣减金额业务：

8.启动 account 项目进行测试

按顺序启动项目：

Eureka
Easy Id Generator
Account
Order
account 项目启动时，会立即从 Rocketmq 收到消息，执行账户扣减业务：

版权说明 : 本文为转载文章, 版权归原作者所有 版权申明

原文链接 : https://blog.csdn.net/weixin_53244569/article/details/121079339

内容来源于网络，如有侵权，请联系作者删除！

分布式微服务spring cloudrocket mq

相关文章

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com