可靠消息最终一致性分布式事务实战_订单微服务业务层实现之发送事务消息-【官方】百战程序员_IT在线教育培训机构


xxxxxxxxxx
/**
     * 添加订单
     * @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接口实现


xxxxxxxxxx
package com.itbaizhan.order.service.impl;

import com.alibaba.fastjson.JSONObject;
import com.itbaizhan.order.entity.Order;
import com.itbaizhan.order.entity.TxLog;
import com.itbaizhan.order.mapper.OrderMapper;
import com.itbaizhan.order.mapper.TxLogMapper;
import com.itbaizhan.order.service.IOrderService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.itbaizhan.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 itbaizhan
 * @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);
    }


}

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

北京市昌平区回龙观镇南店村综合商业楼2楼226室