分布式事务

简介

实际生产过程中因为硬件故障、网络故障、依赖第三方系统故障给我们带来了很多麻烦。原来的重试补偿功能都是写在各业务模块，增加了很多额外的开发工作量，在加上开发人员水平也不一致也很难全面的考虑各种稳定性问题，所以研发出该模块用于解决微服务业务模块不稳定问题。因业务要求，不允许出现失败回滚场景，该模块只实现了事务补偿。 原理：通过N次重试，跳过执行成功的部分，一直重试失败部分，来达到业务最终执行完成。（N次失败后可以通知人工来进行解决）。

实际场景举例： 用户购买了商品，当微信支付成功后，突然订单模块数据库宕机了。 当数据库修复后，之前丢失的订单能正确处理。

特性

1. 支持同步消息与异步消息
2. 提供注解方法，使用简单，学习成本低
3. 任何需要重试的内容都可以使用，适应性强。

快速上手

分布式事务算是框架里面比较重的一个模块了，新的模式需要引用一个客户端，然后配置job和存储。

1、在pom.xml中引入依赖模块

<!-- 客户端事务集成模块-->
<dependency>
	<groupId>com.hbasesoft.framework</groupId>
	<artifactId>framework-tx-integration</artifactId>
	<version>${project.parent.version}</version>
</dependency>

<!-- 客户端消息的存储-->
<dependency>
	<groupId>com.hbasesoft.framework</groupId>
	<artifactId>framework-tx-server-storage-cassandra</artifactId>
	<version>${project.parent.version}</version>
</dependency>

2、修改application.yml中配置

project: #项目信息
 name: tx-demo-client
    
server.port: 8939
 
spring: #应用配置
  application:
    name: ${project.name} 
  data: 
    cassandra: #使用cassandra作为事务消息的存储
      keyspace-name: sgp_tx
      contactPoints: 127.0.0.1
      port: 9042
    
job:
  register:
    url: localhost:2181
    namespace: ${project.name}
  event:
    enable: false

3、 在服务的入口增加@Tx注解, 调用远程接口的地方用TxInvokerProxy.invoke(String, TxInvoker)方法进行包裹，这样就可以保证重试的时候跳过已经执行的代码

TestProducter.java

@GetMapping
@Tx  // 分布式事务注解，里面支持maxRetryTimes 重试次数（默认重试5次）和retryConfigs重试配置的设置
public synchronized String test(final @RequestParam("id") String id) {

   // feClient2Consumer.test为远程方法，如果client2已经执行，下次重试会跳过该方法
    String value1 = TxInvokerProxy.invoke("client2", () -> {
        return feClient2Consumer.test(id);
    });
    System.out.println(value1);

   // feClient3Consumer.test为远程方法
    String value2 = TxInvokerProxy.invoke("client3", () -> {
        return feClient3Consumer.test(id);
    });
    System.out.println(value2);

 // 创造随机失败场景
    if (new Random().nextInt(NUM_5) == 1) {
        throw new RuntimeException();
    }
    System.out.println(i++ + ":" + id);

    return new StringBuilder().append(i).append("client1").append(id).append(':').append(value1).append(':')
        .append(value2).toString();
}

异步消息场景

针对异步消息framework-message模块，我们增加一些api来支撑这个场景，使用起来很方便

1、在pom.xml中引入额外的依赖模块

<dependency>
	<groupId>com.hbasesoft.framework</groupId>
	<artifactId>framework-message-tx</artifactId>
	<version>${project.parent.version}</version>
</dependency>

2、消费者需要继承com.hbasesoft.framework.message.tx.TxEventLinsener， 该接口是EventLinsener的子类，使用方式完全和EventLinsener一样。

@Component
public class TestEventLinsener implements TxEventLinsener {
    private static final int NUM_5 = 5;
    private int i = 0;

    @Override
    public String[] events() {
        return new String[] {
            "testEvent"
        };
    }

    @Override
    public void onEmmit(final String event, final EventData data) {
        // 创造随机失败场景
        if (new Random().nextInt(NUM_5) == 1) {
            throw new RuntimeException();
        }
        System.out.println(i++ + ":" + data.getMsgId());
    }
}

3、生产者使用com.hbasesoft.framework.message.tx.TxEventEmmiter.emmit(String)， 当然为了防止重试后消息重复发送，也是需要用TxInvokerProxy.invoke方法进行包裹

@GetMapping
@Tx  // 分布式事务注解，里面支持maxRetryTimes 重试次数（默认重试5次）和retryConfigs重试配置的设置
public synchronized String test(final @RequestParam("id") String id) {
    // 包裹后，下次重试不会再次触发事件发送
    TxInvokerProxy.invoke("client1", () -> {
        TxEventEmmiter.emmit("testEvent");    // 事务消息的发送方式
        return null;
    });
    System.out.println("emmit event");

   // feClient2Consumer.test为远程方法，如果client2已经执行，下次重试会跳过该方法
    String value1 = TxInvokerProxy.invoke("client2", () -> {
        return feClient2Consumer.test(id);
    });
    System.out.println(value1);

   // feClient3Consumer.test为远程方法
    String value2 = TxInvokerProxy.invoke("client3", () -> {
        return feClient3Consumer.test(id);
    });
    System.out.println(value2);

 // 创造随机失败场景
    if (new Random().nextInt(NUM_5) == 1) {
        throw new RuntimeException();
    }
    System.out.println(i++ + ":" + id);

    return new StringBuilder().append(i).append("client1").append(id).append(':').append(value1).append(':')
        .append(value2).toString();
}

配置参数说明