RocketMQ提供了事务消息—->高可靠消息
首先需要搭建RocketMQ服务器:文章地址

导入无事务项目

  1. 将项目解压至rocketmq-dtx工程目录
  2. 导入
    • 添加模块
    • 导入pom文件导入模块

订单服务使用事务消息异步调用账户服务

添加事务消息

步骤如下:

  1. 父项目添加spring rocketMQ依赖

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

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

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

  3. 添加新的数据表tx_table记录事务执行状态

修改 db-init 项目中的 order.sql 文件，创建 tx_table 表
order_rocketmq.sql

  1. 添加实体类TXInfo与Mapper

    1. @Data
    2. @NoArgsConstructor
    3. @AllArgsConstructor
    4. @Accessors(chain = true)
    5. public class TxInfo {
    6.  /**
    7.   * 事务id
    8.   */
    9.  private String xid;
    10.  /**
    11.   * 事务执行状态,0成功,1失败,2未知
    12.   */
    13.  private Integer status;
    14.  /**
    15.   * 存储时间
    16.   */
    17.  private Long created;
    18. }

    mapper:

    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.entity.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>
    14.  <select id="selectById" resultMap="BaseResultMap">
    15.      SELECT `xid`,`created_at`,`status` FROM tx_table WHERE xid=#{xid};
    16.  </select>
    17. </mapper>

  2. 创建AccountMessage,封装发给用户的调用消息:UserId,money,txId

    1. @Data
    2. @NoArgsConstructor
    3. @AllArgsConstructor
    4. @Accessors(chain = true)
    5. public class AccountMessage {
    6.  private Long userId;
    7.  private BigDecimal money;
    8.  private String xid;
    9. }

  3. 将封装的数据转换为Json字符串使用JsonUtil

    1. @Slf4j
    2. public class JsonUtil {
    3.  private static ObjectMapper mapper;
    4.  private static JsonInclude.Include DEFAULT_PROPERTY_INCLUSION = JsonInclude.Include.NON_DEFAULT;
    5.  private static boolean IS_ENABLE_INDENT_OUTPUT = false;
    6.  private static String CSV_DEFAULT_COLUMN_SEPARATOR = ",";
    7.  static {
    8.      try {
    9.          initMapper();
    10.          configPropertyInclusion();
    11.          configIndentOutput();
    12.          configCommon();
    13.      } catch (Exception e) {
    14.          log.error("jackson config error", e);
    15.      }
    16.  }
    18.  private static void initMapper() {
    19.      mapper = new ObjectMapper();
    20.  }
    22.  private static void configCommon() {
    23.      config(mapper);
    24.  }
    26.  private static void configPropertyInclusion() {
    27.      mapper.setSerializationInclusion(DEFAULT_PROPERTY_INCLUSION);
    28.  }
    30.  private static void configIndentOutput() {
    31.      mapper.configure(SerializationFeature.INDENT_OUTPUT, IS_ENABLE_INDENT_OUTPUT);
    32.  }
    34.  private static void config(ObjectMapper objectMapper) {
    35.      objectMapper.enable(JsonGenerator.Feature.WRITE_BIGDECIMAL_AS_PLAIN);
    36.      objectMapper.enable(DeserializationFeature.ACCEPT_EMPTY_STRING_AS_NULL_OBJECT);
    37.      objectMapper.enable(DeserializationFeature.ACCEPT_SINGLE_VALUE_AS_ARRAY);
    38.      objectMapper.enable(DeserializationFeature.FAIL_ON_READING_DUP_TREE_KEY);
    39.      objectMapper.enable(DeserializationFeature.FAIL_ON_NUMBERS_FOR_ENUMS);
    40.      objectMapper.disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES);
    41.      objectMapper.disable(DeserializationFeature.FAIL_ON_NULL_FOR_PRIMITIVES);
    42.      objectMapper.disable(SerializationFeature.FAIL_ON_EMPTY_BEANS);
    43.      objectMapper.enable(JsonParser.Feature.ALLOW_COMMENTS);
    44.      objectMapper.disable(JsonGenerator.Feature.ESCAPE_NON_ASCII);
    45.      objectMapper.enable(JsonGenerator.Feature.IGNORE_UNKNOWN);
    46.      objectMapper.enable(JsonParser.Feature.ALLOW_UNQUOTED_FIELD_NAMES);
    47.      objectMapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
    48.      objectMapper.setDateFormat(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"));
    49.      objectMapper.enable(JsonParser.Feature.ALLOW_SINGLE_QUOTES);
    50.      objectMapper.registerModule(new ParameterNamesModule());
    51.      objectMapper.registerModule(new Jdk8Module());
    52.      objectMapper.registerModule(new JavaTimeModule());
    53.  }
    54.  public static void setSerializationInclusion(JsonInclude.Include inclusion) {
    55.      DEFAULT_PROPERTY_INCLUSION = inclusion;
    56.      configPropertyInclusion();
    57.  }
    59.  public static void setIndentOutput(boolean isEnable) {
    60.      IS_ENABLE_INDENT_OUTPUT = isEnable;
    61.      configIndentOutput();
    62.  }
    64.  public static <V> V from(URL url, Class<V> c) {
    65.      try {
    66.          return mapper.readValue(url, c);
    67.      } catch (IOException e) {
    68.          log.error("jackson from error, url: {}, type: {}", url.getPath(), c, e);
    69.          return null;
    70.      }
    71.  }
    73.  public static <V> V from(InputStream inputStream, Class<V> c) {
    74.      try {
    75.          return mapper.readValue(inputStream, c);
    76.      } catch (IOException e) {
    77.          log.error("jackson from error, type: {}", c, e);
    78.          return null;
    79.      }
    80.  }
    82.  public static <V> V from(File file, Class<V> c) {
    83.      try {
    84.          return mapper.readValue(file, c);
    85.      } catch (IOException e) {
    86.          log.error("jackson from error, file path: {}, type: {}", file.getPath(), c, e);
    87.          return null;
    88.      }
    89.  }
    91.  public static <V> V from(Object jsonObj, Class<V> c) {
    92.      try {
    93.          return mapper.readValue(jsonObj.toString(), c);
    94.      } catch (IOException e) {
    95.          log.error("jackson from error, json: {}, type: {}", jsonObj.toString(), c, e);
    96.          return null;
    97.      }
    98.  }
    100.  public static <V> V from(String json, Class<V> c) {
    101.      try {
    102.          return mapper.readValue(json, c);
    103.      } catch (IOException e) {
    104.          log.error("jackson from error, json: {}, type: {}", json, c, e);
    105.          return null;
    106.      }
    107.  }
    109.  public static <V> V from(URL url, TypeReference<V> type) {
    110.      try {
    111.          return mapper.readValue(url, type);
    112.      } catch (IOException e) {
    113.          log.error("jackson from error, url: {}, type: {}", url.getPath(), type, e);
    114.          return null;
    115.      }
    116.  }
    118.  public static <V> V from(InputStream inputStream, TypeReference<V> type) {
    119.      try {
    120.          return mapper.readValue(inputStream, type);
    121.      } catch (IOException e) {
    122.          log.error("jackson from error, type: {}", type, e);
    123.          return null;
    124.      }
    125.  }
    127.  public static <V> V from(File file, TypeReference<V> type) {
    128.      try {
    129.          return mapper.readValue(file, type);
    130.      } catch (IOException e) {
    131.          log.error("jackson from error, file path: {}, type: {}", file.getPath(), type, e);
    132.          return null;
    133.      }
    134.  }
    136.  public static <V> V from(Object jsonObj, TypeReference<V> type) {
    137.      try {
    138.          return mapper.readValue(jsonObj.toString(), type);
    139.      } catch (IOException e) {
    140.          log.error("jackson from error, json: {}, type: {}", jsonObj.toString(), type, e);
    141.          return null;
    142.      }
    143.  }
    145.  public static <V> V from(String json, TypeReference<V> type) {
    146.      try {
    147.          return mapper.readValue(json, type);
    148.      } catch (IOException e) {
    149.          log.error("jackson from error, json: {}, type: {}", json, type, e);
    150.          return null;
    151.      }
    152.  }
    154.  public static <V> String to(List<V> list) {
    155.      try {
    156.          return mapper.writeValueAsString(list);
    157.      } catch (JsonProcessingException e) {
    158.          log.error("jackson to error, obj: {}", list, e);
    159.          return null;
    160.      }
    161.  }
    163.  public static <V> String to(V v) {
    164.      try {
    165.          return mapper.writeValueAsString(v);
    166.      } catch (JsonProcessingException e) {
    167.          log.error("jackson to error, obj: {}", v, e);
    168.          return null;
    169.      }
    170.  }
    172.  public static <V> void toFile(String path, List<V> list) {
    173.      try (Writer writer = new FileWriter(new File(path), true)) {
    174.          mapper.writer().writeValues(writer).writeAll(list);
    175.          writer.flush();
    176.      } catch (Exception e) {
    177.          log.error("jackson to file error, path: {}, list: {}", path, list, e);
    178.      }
    179.  }
    181.  public static <V> void toFile(String path, V v) {
    182.      try (Writer writer = new FileWriter(new File(path), true)) {
    183.          mapper.writer().writeValues(writer).write(v);
    184.          writer.flush();
    185.      } catch (Exception e) {
    186.          log.error("jackson to file error, path: {}, obj: {}", path, v, e);
    187.      }
    188.  }
    190.  public static String getString(String json, String key) {
    191.      if (StringUtils.isEmpty(json)) {
    192.          return null;
    193.      }
    194.      try {
    195.          JsonNode node = mapper.readTree(json);
    196.          if (null != node) {
    197.              return node.get(key).asText();
    198.          } else {
    199.              return null;
    200.          }
    201.      } catch (IOException e) {
    202.          log.error("jackson get string error, json: {}, key: {}", json, key, e);
    203.          return null;
    204.      }
    205.  }
    207.  public static Integer getInt(String json, String key) {
    208.      if (StringUtils.isEmpty(json)) {
    209.          return null;
    210.      }
    211.      try {
    212.          JsonNode node = mapper.readTree(json);
    213.          if (null != node) {
    214.              return node.get(key).intValue();
    215.          } else {
    216.              return null;
    217.          }
    218.      } catch (IOException e) {
    219.          log.error("jackson get int error, json: {}, key: {}", json, key, e);
    220.          return null;
    221.      }
    222.  }
    224.  public static Long getLong(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).longValue();
    232.          } else {
    233.              return null;
    234.          }
    235.      } catch (IOException e) {
    236.          log.error("jackson get long error, json: {}, key: {}", json, key, e);
    237.          return null;
    238.      }
    239.  }
    241.  public static Double getDouble(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).doubleValue();
    249.          } else {
    250.              return null;
    251.          }
    252.      } catch (IOException e) {
    253.          log.error("jackson get double error, json: {}, key: {}", json, key, e);
    254.          return null;
    255.      }
    256.  }
    258.  public static BigInteger getBigInteger(String json, String key) {
    259.      if (StringUtils.isEmpty(json)) {
    260.          return new BigInteger(String.valueOf(0.00));
    261.      }
    262.      try {
    263.          JsonNode node = mapper.readTree(json);
    264.          if (null != node) {
    265.              return node.get(key).bigIntegerValue();
    266.          } else {
    267.              return null;
    268.          }
    269.      } catch (IOException e) {
    270.          log.error("jackson get biginteger error, json: {}, key: {}", json, key, e);
    271.          return null;
    272.      }
    273.  }
    275.  public static BigDecimal getBigDecimal(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).decimalValue();
    283.          } else {
    284.              return null;
    285.          }
    286.      } catch (IOException e) {
    287.          log.error("jackson get bigdecimal error, json: {}, key: {}", json, key, e);
    288.          return null;
    289.      }
    290.  }
    292.  public static boolean getBoolean(String json, String key) {
    293.      if (StringUtils.isEmpty(json)) {
    294.          return false;
    295.      }
    296.      try {
    297.          JsonNode node = mapper.readTree(json);
    298.          if (null != node) {
    299.              return node.get(key).booleanValue();
    300.          } else {
    301.              return false;
    302.          }
    303.      } catch (IOException e) {
    304.          log.error("jackson get boolean error, json: {}, key: {}", json, key, e);
    305.          return false;
    306.      }
    307.  }
    309.  public static byte[] getByte(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).binaryValue();
    317.          } else {
    318.              return null;
    319.          }
    320.      } catch (IOException e) {
    321.          log.error("jackson get byte error, json: {}, key: {}", json, key, e);
    322.          return null;
    323.      }
    324.  }
    326.  public static <T> ArrayList<T> getList(String json, String key) {
    327.      if (StringUtils.isEmpty(json)) {
    328.          return null;
    329.      }
    330.      String string = getString(json, key);
    331.      return from(string, new TypeReference<ArrayList<T>>() {});
    332.  }
    334.  public static <T> String add(String json, String key, T value) {
    335.      try {
    336.          JsonNode node = mapper.readTree(json);
    337.          add(node, key, value);
    338.          return node.toString();
    339.      } catch (IOException e) {
    340.          log.error("jackson add error, json: {}, key: {}, value: {}", json, key, value, e);
    341.          return json;
    342.      }
    343.  }
    345.  private static <T> void add(JsonNode jsonNode, String key, T value) {
    346.      if (value instanceof String) {
    347.          ((ObjectNode) jsonNode).put(key, (String) value);
    348.      } else if (value instanceof Short) {
    349.          ((ObjectNode) jsonNode).put(key, (Short) value);
    350.      } else if (value instanceof Integer) {
    351.          ((ObjectNode) jsonNode).put(key, (Integer) value);
    352.      } else if (value instanceof Long) {
    353.          ((ObjectNode) jsonNode).put(key, (Long) value);
    354.      } else if (value instanceof Float) {
    355.          ((ObjectNode) jsonNode).put(key, (Float) value);
    356.      } else if (value instanceof Double) {
    357.          ((ObjectNode) jsonNode).put(key, (Double) value);
    358.      } else if (value instanceof BigDecimal) {
    359.          ((ObjectNode) jsonNode).put(key, (BigDecimal) value);
    360.      } else if (value instanceof BigInteger) {
    361.          ((ObjectNode) jsonNode).put(key, (BigInteger) value);
    362.      } else if (value instanceof Boolean) {
    363.          ((ObjectNode) jsonNode).put(key, (Boolean) value);
    364.      } else if (value instanceof byte[]) {
    365.          ((ObjectNode) jsonNode).put(key, (byte[]) value);
    366.      } else {
    367.          ((ObjectNode) jsonNode).put(key, to(value));
    368.      }
    369.  }
    371.  public static String remove(String json, String key) {
    372.      try {
    373.          JsonNode node = mapper.readTree(json);
    374.          ((ObjectNode) node).remove(key);
    375.          return node.toString();
    376.      } catch (IOException e) {
    377.          log.error("jackson remove error, json: {}, key: {}", json, key, e);
    378.          return json;
    379.      }
    380.  }
    382.  public static <T> String update(String json, String key, T value) {
    383.      try {
    384.          JsonNode node = mapper.readTree(json);
    385.          ((ObjectNode) node).remove(key);
    386.          add(node, key, value);
    387.          return node.toString();
    388.      } catch (IOException e) {
    389.          log.error("jackson update error, json: {}, key: {}, value: {}", json, key, value, e);
    390.          return json;
    391.      }
    392.  }
    394.  public static String format(String json) {
    395.      try {
    396.          JsonNode node = mapper.readTree(json);
    397.          return mapper.writerWithDefaultPrettyPrinter().writeValueAsString(node);
    398.      } catch (IOException e) {
    399.          log.error("jackson format json error, json: {}", json, e);
    400.          return json;
    401.      }
    402.  }
    404.  public static boolean isJson(String json) {
    405.      try {
    406.          mapper.readTree(json);
    407.          return true;
    408.      } catch (Exception e) {
    409.          log.error("jackson check json error, json: {}", json, e);
    410.          return false;
    411.      }
    412.  }
    414.  private static InputStream getResourceStream(String name) {
    415.      return JsonUtil.class.getClassLoader().getResourceAsStream(name);
    416.  }
    418.  private static InputStreamReader getResourceReader(InputStream inputStream) {
    419.      if (null == inputStream) {
    420.          return null;
    421.      }
    422.      return new InputStreamReader(inputStream, StandardCharsets.UTF_8);
    423.  }
    424. }

  4. OrderServiceImpl发送事务消息

    1. @Override
    2.  public void createOrder(Order order) {
    3.      //准备消息数据
    4.      String xid = UUID.randomUUID().toString().replace("-", "");
    5.      AccountMessage message = new AccountMessage(order.getUserId(), order.getMoney(), xid);
    6.      String json = JsonUtil.to(message);
    8.      //将json字符串封装至spring的同用message对象
    9.      assert json != null;
    10.      Message<String> msg = MessageBuilder.withPayload(json).build();
    12.      /*参数 1. 消息主题 2. 消息 3. 触发监听器执行业务时需要的业务数据参数*/
    13.      rocketMQTemplate.sendMessageInTransaction("orderTopic", msg, order);
    14.  }

  5. 实现事务监听器

    1. /**
    2.   * 执行本地事务
    3.   *
    4.   * @param message 消息
    5.   * @param o       参数
    6.   * @return 状态
    7.   */
    8.  @Transactional
    9.  @Override
    10.  public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
    11.      //用于返回状态
    12.      RocketMQLocalTransactionState state;
    13.      //用于在数据库中保存状态
    14.      Integer status;
    15.      try {
    16.          doCreateOrder((Order) o);
    17.          state = RocketMQLocalTransactionState.COMMIT;
    18.          status = 0;
    19.      } catch (Exception e) {
    20.          log.error("创建订单失败", e);
    21.          state = RocketMQLocalTransactionState.ROLLBACK;
    22.          status = 1;
    23.      }
    24.      String json = new String((byte[]) message.getPayload());
    25.      String xid = JsonUtil.getString(json, "xid");
    26.      txMapper.insert(new TxInfo(xid, status, System.currentTimeMillis()));
    27.      return state;
    28.  }
    30.  /**
    31.   * 处理事务回查
    32.   *
    33.   * @param message 消息
    34.   * @return 状态
    35.   */
    36.  @Override
    37.  public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
    38.      String json = new String((byte[]) message.getPayload());
    39.      String xid = JsonUtil.getString(json, "xid");
    40.      TxInfo txInfo = txMapper.selectById(xid);
    41.      if (txInfo == null) {
    42.          return RocketMQLocalTransactionState.UNKNOWN;
    43.      }
    44.      switch (txInfo.getStatus()) {
    45.          case 0:
    46.              return RocketMQLocalTransactionState.COMMIT;
    47.          case 1:
    48.              return RocketMQLocalTransactionState.ROLLBACK;
    49.          default:
    50.              return RocketMQLocalTransactionState.UNKNOWN;
    51.      }
    52.  }

    账户接受事务消息,扣减账户

  6. 添加依赖

  7. 修改yml配置,添加rocketMQ的name server连接地址
  8. 创建消息对象AccountMessage
  9. JsonUtil
  10. 新建消费者类,AccountConsumer,实现消费者接口
  11. 通过注解配置接收消息

  12. 扣减账户

    1. @Component
    2. @RocketMQMessageListener(topic = "orderTopic", consumerGroup = "account-consumer")
    3. public class AccountConsumer implements RocketMQListener<String> {
    4.  private final AccountService accountService;
    6.  @Autowired
    7.  public AccountConsumer(AccountService accountService) {
    8.      this.accountService = accountService;
    9.  }
    11.  @Override
    12.  public void onMessage(String json) {
    13.      AccountMessage accountMessage = JsonUtil.from(json, AccountMessage.class);
    14.      assert accountMessage != null;
    15.      accountService.decrease(accountMessage.getUserId(), accountMessage.getMoney());
    16.  }
    17. }