[学习交流] 【西安校区】分布式事务——纯MQ实现

本帖最后由逆风TO 于 2020-3-4 10:22 编辑

一、MQ实现分布式事务，最简单的原理框架：
借助MQ的消息可靠传递，实现业务间解耦、事务强一致

1、>> 生产者发送消息做可靠性检查，确保消息真正投递出去；

2、>> 消费者做幂等，确保业务没有重复执行；

3、>> 消费者做异常重试，反复出错时需要捕捉异常并记录，以便手工干预；

二、场景实践：
场景
以支付宝转账到余额宝为例，在支付宝已经扣款成功的情况下，余额宝一定收到转账

>> 支付宝和余额宝是两个微服务；

>> 用户用支付宝转1万元到余额宝；

支付宝账户先扣除金额，MQ通知余额宝账户添加金额；

>> 支付宝账户表: update A set amount = amount - 10000 where userId = 1;

>> 余额宝账户表: update B set amount = amount + 10000 where userId = 1;

具体操作
1、创建队列
>> 创建一个持久化的队列，名称为money；

Durable（持久化保存），Transient（即时保存）。

2、 maven引入
（余额宝和支付宝两个服务都需要）

[Java] 纯文本查看 复制代码

<dependency>

    <groupId>org.springframework.boot</groupId>

    <artifactId>spring-boot-starter-amqp</artifactId>

</dependency>

AMQP，即Advanced Message Queuing Protocol，一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同的开发语言等条件的限制。

3、添加配置
（余额宝和支付宝两个服务都需要）

[Java] 纯文本查看 复制代码

spring:    

    rabbitmq:

        host: 192.168.222.26 #MQ地址

        port: 5672 #MQ端口

        username: admin #MQ用户名

        password: admin #MQ密码

        publisher-confirms: true #开启消息发送成功监听

        publisher-returns: true #开启消息发送失败监听

        listener:

            simple:

                acknowledge-mode: manual #手动提交事务

4 支付宝端代码编写
向支付宝账号扣款，同时发消息到队列中，通知余额宝账号金额更新

>> 配置MQ发送过程监听

[Java] 纯文本查看 复制代码

package cn.my.server.clientdemo.zhifu;
 
import javax.annotation.PostConstruct;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
/**
 * 配置MQ发送过程监听
 * @author twotiger-wxm.
 * @date 2020-3-3.
 */
@Configuration
public class RabbmitMqConfig {
 
        @Autowired
        private RabbitTemplate rabbitTemplate;
 
        @PostConstruct
        public void init() {
            // 设置发送成功回调
            rabbitTemplate.setConfirmCallback(initConfirmCallback());
            // 设置发送失败回调
            rabbitTemplate.setReturnCallback(initReturnCallback());
        }
 
        @Bean
        public ReturnCallback initReturnCallback(){
            return new FailedListener();
        }
 
        @Bean
        public ConfirmCallback initConfirmCallback(){
            return new SuccessListener();
        }
 
}

[AppleScript] 纯文本查看 复制代码

package cn.my.server.clientdemo.zhifu;

import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback;

import org.springframework.amqp.rabbit.support.CorrelationData;

/**

 * 消息发送成功时触发

 * 测试发现队列不存在也会触发，但会在FailedListener之后触发

 */

public class SuccessListener implements ConfirmCallback{

    @Override

    public void confirm(CorrelationData data, boolean success, String result) {

        System.out.println(data+","+success+","+result);

        if(null!=data){

            System.out.println("业务主键："+data.getId());

        }

        if(success){   

            boolean resultT = true;// 获取流水表状态，为fail则失败（因为队列不存在也会触发，且在失败监听之后触发，所以此处优先排除一下失败监听的操作。）

            if(resultT){

                //更新支付宝账户金额

                System.out.println("消息发送到MQ成功");

                System.out.println("【支付宝账户扣款】  update A set amount = amount - 10000 where userId = 1");

            }else{

                System.out.println("消息发送到MQ失败");

                System.out.println("【支付宝账户不扣款】根据业务主键更新支付宝流水表状态为失败,update A_MESSAGE set sts = fail where transId=000001");

            }      

        }else{

            System.out.println("消息发送到MQ失败原因："+result);

            System.out.println("【支付宝账户不扣款】根据业务主键更新支付宝流水表状态为失败,update A_MESSAGE set sts = fail where transId=000001");

        }      

    }

     

} 

package cn.my.server.clientdemo.zhifu;

import java.util.HashMap;

import org.springframework.amqp.core.Message;

import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;

import com.fasterxml.jackson.databind.ObjectMapper;

/**

 * 消息发送失败时，监听被触发

 * 一般像队列不存在,或者网络中断时会触发，且第一个触发

 */

public class FailedListener implements ReturnCallback{

    private ObjectMapper mapper = new ObjectMapper();

    @Override

    public void returnedMessage(Message message, int state, String result, String arg3, String queneName) {

        System.out.println(message+","+state+","+result+","+arg3+","+queneName);

         

        try {

            String msg = new String(message.getBody());

            HashMap data = mapper.readValue(msg, HashMap.class);// 失败的消息内容

            System.out.println("失败的转账记录："+mapper.writeValueAsString(data));

            System.out.println("【支付宝账户不扣款】根据业务主键更新支付宝流水表状态为失败,update A_MESSAGE set sts = fail where transId=000001");

        } catch (Exception e) {

            e.printStackTrace();

        }

    } 

     

}

>> 转账接口编写

[Java] 纯文本查看 复制代码

package cn.my.server.clientdemo.zhifu;

import java.util.HashMap;

import java.util.Map;

import org.springframework.amqp.AmqpException;

import org.springframework.amqp.rabbit.core.RabbitTemplate;

import org.springframework.amqp.rabbit.support.CorrelationData;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.transaction.annotation.Transactional;

import org.springframework.web.bind.annotation.GetMapping;

import org.springframework.web.bind.annotation.RequestMapping;

import org.springframework.web.bind.annotation.RestController;

import com.fasterxml.jackson.core.JsonProcessingException;

import com.fasterxml.jackson.databind.ObjectMapper;

/**

 * 支付宝端业务测试

 */

@RestController

@RequestMapping("/api/test")

public class ZhiFuBaoResource {

    /**

     * 测试接口

     * @throws AmqpException

     * @throws JsonProcessingException

     */

    @GetMapping("/mq")

    public void trans1() throws AmqpException, JsonProcessingException {

        trans();

    }

    @Autowired

    private RabbitTemplate rabbitTemplate;

    private ObjectMapper mapper = new ObjectMapper();

    /**

     * 转账方法

     * @return

     * @throws AmqpException

     * @throws JsonProcessingException

     */

    @Transactional

    public boolean trans() throws AmqpException, JsonProcessingException {

        Map data = new HashMap<String, Object>();

        String transId = "00000001";//生成业务流水号

        data.put("transId", transId);

        data.put("userId", "1");

        data.put("money", 10000);

         

        //A_MESSAGE交易流水表,主键是【交易编号】，重复插入会报异常，类似于幂等操作

        System.out.println("【支付宝记账操作】重复插入会报异常，类似于幂等操作  insert into A_MESSAGE(transId,userId,money) values (000001,1,10000)");

             

        // 流水号传入，方便消息发送失败时做操作

        CorrelationData correlationData = new CorrelationData(transId);

        rabbitTemplate.convertAndSend("","money",mapper.writeValueAsString(data),correlationData);

        return true;

    }

}

5 余额宝端代码编写

监听支付宝发到消息队列中的消息，做余额宝账号金额更新

[Java] 纯文本查看 复制代码

package cn.my.server.clientdemo.yuer;

import java.io.IOException;

import java.util.Collections;

import java.util.HashMap;

import java.util.Map;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.amqp.core.Message;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;

import org.springframework.amqp.rabbit.annotation.RabbitListener;

import org.springframework.retry.RecoveryCallback;

import org.springframework.retry.RetryCallback;

import org.springframework.retry.RetryContext;

import org.springframework.retry.backoff.FixedBackOffPolicy;

import org.springframework.retry.policy.SimpleRetryPolicy;

import org.springframework.retry.support.RetryTemplate;

import org.springframework.stereotype.Component;

import org.springframework.transaction.annotation.Transactional;

import com.fasterxml.jackson.core.JsonProcessingException;

import com.fasterxml.jackson.databind.ObjectMapper;

import com.rabbitmq.client.Channel;

/**

 * 余额宝端消息监听

 */

@Component

public class YuErBaoMessageListeners {

    private final Logger log = LoggerFactory.getLogger(this.getClass());

    private ObjectMapper mapper = new ObjectMapper();

 

    /**

     * 监听消息队列

     * @param message  消息内容

     * @param channel  消息渠道

     * @throws IOException  异常

     */

    @RabbitListener(queues = "money")

    @RabbitHandler

    public void receiveQueue(Message message, Channel channel) throws IOException {

        String msg = "";

        try {

            // 业务处理逻辑

            msg = new String(message.getBody());

            Map data = mapper.readValue(msg, HashMap.class);

            retry(data);

            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);// 手动应答消息已经处理

        } catch (Exception e) {

            log.error("MQ接收消息内容[" + msg + "],后处理异常：" + e);

            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);// 手动应答消息已经处理

        }

    }

 

    /**

     * 更新余额宝账户金额

     *

     * @param map

     */

    @Transactional

    public void bizOp(Map<String, Object> map) {

        // B_MESSAGE交易流水表,主键是【交易编号】，重复插入会报异常，类似于幂等操作

        System.out.println(

            "【余额宝记账操作】重复插入会报异常，类似于幂等操作 insert into B_MESSAGE(transId,userId,money) values (000001,1,10000)");

        // 更新余额宝账户金额

        System.out.println("【余额宝账户入款】 update B set amount = amount +10000 where userId = 1");

 

    }

 

    /**

     * 一直报错，重试次数用完了，保存如下信息供人工干预

     *

     * @param map

     * @throws JsonProcessingException

     */

    public void failed(Map<String, Object> map) throws JsonProcessingException {

        System.out.println("一直报错，重试次数用完了，保存如下信息供人工干预:\n" + mapper.writeValueAsString(map));

    }

 

    /**

     * 异常时最多重试 3次，成功为止

     *

     * @param map

     *            输入参数

     */

    private void retry(Map<String, Object> map) {

        // 构建重试模板实例

        RetryTemplate retryTemplate = new RetryTemplate();

        // 设置重试次数

        SimpleRetryPolicy policy = new SimpleRetryPolicy(3,

                Collections.<Class<? extends Throwable>, Boolean>singletonMap(Exception.class, true));

        // 设置重试间隔时间

        FixedBackOffPolicy fixedBackOffPolicy = new FixedBackOffPolicy();

        fixedBackOffPolicy.setBackOffPeriod(100);

        retryTemplate.setRetryPolicy(policy);

        retryTemplate.setBackOffPolicy(fixedBackOffPolicy);

        // 编写业务处理代码逻辑

        final RetryCallback<Object, Exception> retryCallback = new RetryCallback<Object, Exception>() {

            public Object doWithRetry(RetryContext context) throws Exception {

                System.out.println("第" + (1 + context.getRetryCount()) + "次处理");

                try {

                    bizOp(map);

                } catch (Exception e) {

                    e.printStackTrace();

                    throw new Exception("捕捉到业务处理异常，需要抛出");// 这个点特别注意，重试的根源通过Exception返回

                }

                return null;

            }

        };

        // 重试次数执行完依然报错，走如下逻辑

        final RecoveryCallback<Object> recoveryCallback = new RecoveryCallback<Object>() {

            public Object recover(RetryContext context) throws Exception {

                failed(map);

                return null;

            }

        };

        try {

            // 由retryTemplate 执行execute方法开始逻辑执行

            retryTemplate.execute(retryCallback, recoveryCallback);

        } catch (Exception e) {

            e.printStackTrace();

        }

    }

}

6 测试

>> 浏览器直接访问 /api/test/mq

>> 控制台结果如下：

[Java] 纯文本查看 复制代码

【支付宝记账操作】重复插入会报异常，类似于幂等操作  insert into A_MESSAGE(transId,userId,money) values (000001,1,10000)

【余额宝记账操作】重复插入会报异常，类似于幂等操作 insert into B_MESSAGE(transId,userId,money) values (000001,1,10000)

【余额宝账户入款】 update B set amount = amount + 10000 where userId = 1

【支付宝账户扣款】 update A set amount = amount - 10000 where userId = 1

帐号		自动登录	找回密码
密码			加入黑马

[学习交流] 【西安校区】分布式事务——纯MQ实现

0 个回复