分布式事物【RocketMQ事务消息、Docker安装 RocketMQ、实现订单微服务、订单微服务业务层实现】(八)-全面详解（学习总结---从入门到深化）（上）：https://developer.aliyun.com/article/1419990

编写配置文件

server:
 port: 9090
spring:
 application:
   name: tx-msg-stock
 datasource:
   url: jdbc:mysql://192.168.66.100:3306/txmsg-order?
useUnicode=true&characterEncoding=UTF8&useOldAliasMetadataBehavior=true&autoReconnec
t=true&failOverReadOnly=false&useSSL=false
   username: root
   password: 123456
   driver-class-name: com.mysql.cj.jdbc.Driver
################ RocketMQ 配置 ##########
rocketmq:
 name-server: 192.168.66.100:9876
 producer:
   group: order-group

编写主启动类

/**
 * 订单微服务启动成功
 */
@Slf4j
@MapperScan("com.tong.order.mapper")
@SpringBootApplication
public class OrderMain9090 {
    public static void main(String[] args) {
      SpringApplication.run(OrderMain9090.class,args);
        log.info("************* 订单微服务启动成功*******");
   }
}

代码生成

package com.tong.utils;
import com.baomidou.mybatisplus.generator.FastAutoGenerator;
import com.baomidou.mybatisplus.generator.config.rules.NamingStrategy;
import java.util.Arrays;
import java.util.List;
public class CodeGenerator {
    public static void main(String[] args) {
      FastAutoGenerator.create("jdbc:mysql://192.168.66.102:3306/tx-msg-order", "root", "123456")
               .globalConfig(builder -> {
                    builder.author("tong")// 设置作者
                           .commentDate("MMdd") // 注释日期格式
                           .outputDir(System.getProperty("user.dir")+"/rocketmq-msg/orders"+ "/src/main/java/")
                           .fileOverride(); //覆盖文件
               })
                // 包配置
               .packageConfig(builder -> {
                  builder.parent("com.tong.orders") // 包名前缀
                           .entity("entity")//实体类包名
                           .mapper("mapper")//mapper接口包名
                           .service("service"); //service包名
               })
              .strategyConfig(builder -> {
                    // 设置需要生成的表名
                  builder.addInclude(Arrays.asList("orders","tx_log"))
                            // 开始实体类配置
                           .entityBuilder()
                            // 开启lombok模型
                           .enableLombok()
                            //表名下划线转驼峰
                           .naming(NamingStrategy.underline_to_camel)
                            //列名下划线转驼峰
                           .columnNaming(NamingStrategy.underline_to_camel);
               })
               .execute();
   }
}

创建TxMessage类

在项目的com.itbaizhan.orders.tx包下创建TxMessage类，主要用来封装实现分布式事务时，在订单微服务、RocketMQ消息中间件和库存微服务之间传递的全局事务消息，项目中会通过事务消息实现幂等。

@Data
@NoArgsConstructor
@AllArgsConstructor
public class TxMessage implements Serializable
{
    private static final long serialVersionUID = -4704980150056885074L;
    /**
     * 商品id
     */
    private Long productId;
    /**
     * 商品购买数量
     */
    private Integer payCount;
    /**
     * 全局事务编号
     */
    private String txNo;
}

可靠消息最终一致性分布式事务实战_订单微服务业务层实现

业务逻辑层主要实现了用户提交订单后的业务逻辑。

编写OrderService接口

    /**
     * 添加订单
     * @param productId 商品id
     * @param payCount 购买数量
     */
    void save(Long productId,Integer payCount);
    /**
     * 提交订单同时保存事务信息
     */
    void submitOrderAndSaveTxNo(TxMessage txMessage);
    /**
     * 提交订单
     * @param productId 商品id
     * @param payCount 购买数量
     */
    void submitOrder(Long productId, Integer payCount);

编写OrderService接口实现

package com.itbaizhan.order.service.impl;
import com.alibaba.fastjson.JSONObject;
import com.tong.order.entity.Order;
import com.tong.order.entity.TxLog;
import com.tong.order.mapper.OrderMapper;
import com.tong.order.mapper.TxLogMapper;
import com.tong.order.service.IOrderService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.tong.order.tx.TxMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.Date;
import java.util.UUID;
/**
* <p>
* 服务实现类
* </p>
*
* @author tong
* @since 05-20
*/
@Slf4j
@Service
public class OrderServiceImpl extends
ServiceImpl<OrderMapper, Order> implements IOrderService {
    @Resource
    RocketMQTemplate rocketMQTemplate;
    @Resource
  private TxLogMapper txLogMapper;
    /**
     * 添加
     * @param productId 商品id
     * @param payCount 购买数量
     */
    @Override
    public void save(Long productId, Integer payCount) {
        Order order = new Order();
        // 订单创建时间
        order.setCreateTime(LocalDateTime.now());
        // 生产订单编号
        order.setOrderNo(UUID.randomUUID().toString().replace("-",""));
        // 商品id
        order.setProductId(productId);
        // 购买数量
        order.setPayCount(payCount);
        baseMapper.insert(order);
   }
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void submitOrderAndSaveTxNo(TxMessage txMessage) {
        TxLog txLog = txLogMapper.selectById(txMessage.getTxNo());
    if(txLog != null){
            log.info("订单微服务已经执行过事务,商品id为:{}，事务编号为:{}",txMessage.getProductId(),txMessage.getTxNo());
            return;
       }
        //生成订单
      this.save(txMessage.getProductId(),txMessage.getPayCount());
        //生成订单
        txLog = new TxLog();
        txLog.setTxNo(txMessage.getTxNo());
        txLog.setCreateTime(LocalDateTime.now());
        //添加事务日志
        txLogMapper.insert(txLog);
   }
    /**
     * 提交订单
     * @param productId 商品id
     * @param payCount 购买数量
     */
    @Override
    public void submitOrder(Long productId,Integer payCount) {
        //生成全局分布式序列号
        String txNo = UUID.randomUUID().toString();
        TxMessage txMessage = new TxMessage(productId, payCount, txNo);
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("txMessage", txMessage);
        Message<String> message = MessageBuilder.withPayload(jsonObject.toJSONString()).build();
        //发送事务消息   且该消息不允许消费   
        tx_order_group: 指定版事务消息组
      rocketMQTemplate.sendMessageInTransaction("tx_order_group", "topic_txmsg", message, null);
   }
}

可靠消息最终一致性分布式事务实战_订单微服务监听事务消息

执行本地的业务代码

package com.tong.order.message;
import com.alibaba.fastjson.JSONObject;
import com.tong.order.entity.TxLog;
import com.tong.order.mapper.TxLogMapper;
import com.tong.order.service.IOrderService;
import com.tong.order.service.ITxLogService;
import com.tong.order.tx.TxMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import javax.annotation.Resource;
/**
* @author tong
* @version 1.0.0
* @description 监听事务消息
*/
@Slf4j
@Component
@RocketMQTransactionListener(txProducerGroup = "tx_order_group")
public class OrderTxMessageListener implements
RocketMQLocalTransactionListener {
    @Autowired
    private IOrderService orderService;
    @Resource
    private TxLogMapper txLogMapper;
    /**
     * RocketMQ的Producer本地事务：先执行本地的业务代码（使用Spring的事件管理），判断是否成功。
     * 成功返回： RocketMQLocalTransactionState.COMMIT,
     * 失败返回：RocketMQLocalTransactionState.ROLLBACK
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object obj) 
      {
        try {
            log.info("订单微服务执行本地事务");
            TxMessage txMessage = this.getTxMessage(msg);
            //执行本地事务
            orderService.submitOrderAndSaveTxNo(txMessage);
            //提交事务
            log.info("订单微服务提交事务");
            // COMMIT:即生产者通知Rocket该消息可以消费
            return RocketMQLocalTransactionState.COMMIT;
       } catch (Exception e) {
            e.printStackTrace();
            //异常回滚事务
            log.info("订单微服务回滚事务");
            // ROLLBACK:即生产者通知Rocket将该消息删除
            return RocketMQLocalTransactionState.ROLLBACK;
       }
   }
    private TxMessage getTxMessage(Message msg)
     {
        String messageString = new String((byte[]) msg.getPayload());
        JSONObject jsonObject = JSONObject.parseObject(messageString);
        String txStr = jsonObject.getString("txMessage");
        return JSONObject.parseObject(txStr,TxMessage.class);
   }
}

网络异常消息处理

    /**
     * 因为网络异常或其他原因时，RocketMQ的消息状态处于UNKNOWN时，调用该方法检查Producer的本地 
     * 事务是否已经执行成功,
     * 成功返回： RocketMQLocalTransactionState.COMMIT,
     * 失败返回：RocketMQLocalTransactionState.ROLLBACK
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        log.info("订单微服务查询本地事务");
        TxMessage txMessage = this.getTxMessage(msg);
        // 获取订单的消息
        Integer exists = txLogService.isExistsTx(txMessage.getTxNo());
        if (exists != null) {
            // COMMIT:即生产者通知Rocket该消息可以消费
            return RocketMQLocalTransactionState.COMMIT;
       }
        // UNKNOWN:即生产者通知Rocket继续查询该消息的状态
        return RocketMQLocalTransactionState.UNKNOWN;
}
   private TxMessage getTxMessage(Message msg)
     {
        String messageString = new String((byte[]) msg.getPayload());
        JSONObject jsonObject = JSONObject.parseObject(messageString);
        String txStr = jsonObject.getString("txMessage");
        return JSONObject.parseObject(txStr,TxMessage.class);
     }

可靠消息最终一致性分布式事务实战_实现库存微服务

创建库存微服务tx-msg-stock

引入依赖

   <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starterweb</artifactId>
        </dependency>
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connectorjava</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-bootstarter</artifactId>
            <version>2.0.1</version>
        </dependency>
        <dependency>
            <groupId>com.baomidou</groupId>
            <artifactId>mybatis-plus-bootstarter</artifactId>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
    </dependencies>

编写配置文件

server:
 port: 6060
spring:
 application:
   name: tx-msg-stock
 datasource:
   url: jdbc:mysql://192.168.66.100:3306/txmsg-stock?
useUnicode=true&characterEncoding=UTF8&useOldAliasMetadataBehavior=true&autoReconnec
t=true&failOverReadOnly=false&useSSL=false
   username: root
   password01: 123456
   driver-class-name: com.mysql.cj.jdbc.Driver
################ RocketMQ 配置 ##########
rocketmq:
 name-server: 192.168.66.100:9876

编写主启动类

package com.tong.stock;
import lombok.extern.slf4j.Slf4j;
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
/**
* @author tong
* @version 1.0.0
* @description 库存微服务启动类
*/
@MapperScan("com.tong.stock.mapper")
@Slf4j
@SpringBootApplication
public class StockServerStarter {
    public static void main(String[] args) {
      SpringApplication.run(StockServerStarter.class, args);
        log.info("**************** 库存服务启动成功 ***********");
   }
}

代码生成

package com.tong.utils;
import com.baomidou.mybatisplus.generator.FastAutoGenerator;
import com.baomidou.mybatisplus.generator.config.rules.NamingStrategy;
import java.util.Arrays;
import java.util.List;
public class CodeGenerator {
    public static void main(String[] args) {
      FastAutoGenerator.create("jdbc:mysql://192.168.66.102:3306/tx-msg-stock", "root", "123456")
               .globalConfig(builder -> { builder.author("tong")// 设置作者
                     .commentDate("MMdd") // 注释日期格式
                     .outputDir(System.getProperty("user.dir") +"/rocketmq-msg/stock"+ "/src/main/java/")
                      .fileOverride(); //覆盖文件
               })
                // 包配置
               .packageConfig(builder -> {
                  builder.parent("com.tong.stock") // 包名前缀
                           .entity("entity")//实体类包名
                           .mapper("mapper")//mapper接口包名
                           .service("service"); //service包名
               })
               .strategyConfig(builder -> {
                    // 设置需要生成的表名
                  builder.addInclude(Arrays.asList("stock","tx_log"))
                            // 开始实体类配置
                           .entityBuilder()
                            // 开启lombok模型
                           .enableLombok() //表名下划线转驼峰
                           .naming(NamingStrategy.underline_to_camel)
                            //列名下划线转驼峰
                           .columnNaming(NamingStrategy.underline_to_camel);
               })
               .execute();
   }
}

编写库存接口

public interface StockService {
    /**
     * 根据id查询库存
     * @param id
     * @return
     */
    Stock getStockById(Long id);
    /**
     * 扣减库存
     */
    void decreaseStock(TxMessage txMessage);
}

分布式事物【RocketMQ事务消息、Docker安装 RocketMQ、实现订单微服务、订单微服务业务层实现】(八)-全面详解（学习总结---从入门到深化）（下）

可靠消息最终一致性分布式事务实战_订单微服务业务层实现

可靠消息最终一致性分布式事务实战_订单微服务监听事务消息

可靠消息最终一致性分布式事务实战_实现库存微服务

热门文章

最新文章

相关课程

相关电子书

相关实验场景

推荐镜像