Apache Dubbo .NET实现分布式事务的方式示例

谁践踏了优雅 2023-09-27 17:03 37阅读 0赞

Apache Dubbo .NET 支持多种分布式事务模式，包括 XA、TCC 和 SAGA。以下是使用 TCC 模式实现跨库分布式事务的示例：

如有一个订单服务和一个商品服务，需要在这两个服务之间实现分布式事务。

1.订单服务实现 TCC 接口

订单服务需要实现 TCC 接口，包括 Try、Confirm 和 Cancel 三个方法，分别用于预留库存、提交订单和取消订单。代码示例如下：

public interface OrderService : ITcc
    {
        [TccTryMethod]
        bool TryReserveInventory(Order order);
    
        [TccConfirmMethod]
        bool ConfirmPlaceOrder(Order order);
    
        [TccCancelMethod]
        bool CancelPlaceOrder(Order order);
    }

2.商品服务实现 TCC 接口

商品服务也需要实现 TCC 接口，包括 Try、Confirm 和 Cancel 三个方法，分别用于锁定库存、扣减库存和释放库存。代码示例如下：

public interface ProductService : ITcc
    {
        [TccTryMethod]
        bool TryLockInventory(Order order);
    
        [TccConfirmMethod]
        bool ConfirmDeductInventory(Order order);
    
        [TccCancelMethod]
        bool CancelLockInventory(Order order);
    }

3.客户端调用服务

客户端需要调用订单服务和商品服务的 TCC 接口，使用 Dubbo 提供的 TccTransactionContext 来传递事务上下文信息。代码示例如下：

4.实现查询操作

在实现查询操作时，需要保证查询的数据是事务提交之后的最新状态。为了保证数据的一致性，可以使用 Dubbo 提供的可靠消息机制，在事务提交后发送消息，通知其他服务更新数据。代码示例如下：

var order = orderService.GetOrder(orderId);
    
    if (order != null)
    {
        // 查询商品信息
        var product = productService.GetProduct(order.ProductId);
    
        // 查询日志信息
        var log = logService.GetLog(order.LogId);
    
        // 更新订单状态
        orderService.UpdateOrderStatus(orderId, OrderStatus.Completed);
    
        // 发送消息，通知商品服务和日志服务更新数据
        var message = new OrderCompletedMessage { OrderId = orderId };
        var producer = DubboReferencHelper.GetReference<IProducer>();
        producer.SendMessage(message);
    }

5. 实现 Dubbo 配置 在实现分布式事务时，需要在 Dubbo 配置中添加 TccTransactionFilter 和 TccTransactionManager，以便 Dubbo 在调用 TCC 接口时启动和管理分布式事务。代码示例如下：

<dubbo:consumer filter="tccTransactionFilter" />
    <dubbo:provider filter="tccTransactionFilter" />
    <dubbo:registry address="..." />
    <dubbo:protocol name="dubbo" port="12345" />
    <dubbo:service interface="OrderService" ref="orderServiceImpl" filter="tccTransactionManager" />
    <dubbo:service interface="ProductService" ref="productServiceImpl" filter="tccTransactionManager" />

注意，以上代码示例仅为演示用途，具体实现方式可能因应用场景和具体业务逻辑而异。在实际应用中，还需要考虑容错、重试、性能等方面的问题，以确保系统的可靠性和稳定性。

6.实现 TCC 接口

TCC 接口需要实现 Try、Confirm 和 Cancel 三个方法。Try 方法用于预留资源，Confirm 方法用于确认资源，而 Cancel 方法用于回滚资源。具体实现代码如下：

[Service(interfaceClass = typeof(IProductService), version = "1.0.0")]
    public class ProductServiceImpl : IProductService
    {
        [TccTransaction(confirmMethod = "ConfirmStock", cancelMethod = "CancelStock")]
        public bool ReduceStock(long productId, int amount)
        {
            var product = _productRepository.GetById(productId);
            if (product == null || product.Stock < amount)
            {
                return false;
            }
    
            _productRepository.UpdateStock(productId, product.Stock - amount);
    
            return true;
        }
    
        public void ConfirmStock(long productId, int amount)
        {
            // do nothing
        }
    
        public void CancelStock(long productId, int amount)
        {
            _productRepository.UpdateStock(productId, _productRepository.GetById(productId).Stock + amount);
        }
    }

在以上代码中，ReduceStock 方法用于预留商品库存，如果库存不足则返回 false。ConfirmStock 方法和 CancelStock 方法分别用于确认和回滚预留的库存。当 TCC 事务执行成功时，Dubbo 会调用 Confirm 方法，否则会调用 Cancel 方法。

7.实现 Dubbo 事务管理器

Dubbo 事务管理器用于管理 TCC 事务的生命周期，并在事务发生异常时回滚事务。具体实现代码如下：

public class DubboTransactionManager : ITccTransactionManager
    {
        private readonly ITccTransactionRepository _transactionRepository;
        private readonly ILogger _logger;
    
        public DubboTransactionManager(ITccTransactionRepository transactionRepository, ILogger logger)
        {
            _transactionRepository = transactionRepository;
            _logger = logger;
        }
    
        public void BeginTransaction(string transactionId)
        {
            _transactionRepository.CreateTransaction(transactionId);
        }
    
        public void CommitTransaction(string transactionId)
        {
            _transactionRepository.DeleteTransaction(transactionId);
        }
    
        public void RollbackTransaction(string transactionId)
        {
            var transaction = _transactionRepository.GetTransaction(transactionId);
            if (transaction == null)
            {
                _logger.LogWarning("Transaction {TransactionId} does not exist.", transactionId);
                return;
            }
    
            try
            {
                foreach (var branch in transaction.Branches)
                {
                    branch.Rollback();
                }
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Failed to rollback transaction {TransactionId}.", transactionId);
                throw;
            }
            finally
            {
                _transactionRepository.DeleteTransaction(transactionId);
            }
        }
    }

在以上代码中，BeginTransaction 方法用于创建新的 TCC 事务，CommitTransaction 方法用于提交事务，RollbackTransaction 方法用于回滚事务。

8.实现 Dubbo 事务过滤器

Dubbo 事务过滤器用于拦截 Dubbo 请求，并在需要时启动和管理 TCC 事务。具体实现代码如下：

public class TccTransactionFilter : IFilter
    {
        private readonly ITccTransactionManager _transactionManager;
    
        public TccTransactionFilter(ITccTransactionManager transactionManager)
        {
            _transactionManager = transactionManager;
        }
    
     public async Task<Response> InvokeAsync(IInvocation invocation)
        {
        var method = invocation.MethodInvocationMessage.MethodName;
        var service = invocation.MethodInvocationMessage.TargetInterfaceName;
        if (!TccTransactionAttribute.IsTccTransaction(invocation.MethodInfo))
        {
            return await invoker.InvokeAsync(invocation);
        }
    
        var transactionId = Guid.NewGuid().ToString("N");
    
        _transactionManager.BeginTransaction(transactionId);
    
        try
        {
            // Try phase
            var tryResult = await invoker.InvokeAsync(invocation);
    
            if (!tryResult.IsSuccess)
            {
                _transactionManager.RollbackTransaction(transactionId);
                return tryResult;
            }
    
            // Confirm phase
            _transactionManager.CommitTransaction(transactionId);
    
            return tryResult;
        }
        catch (Exception ex)
        {
            _transactionManager.RollbackTransaction(transactionId);
            throw;
        }
    }

在以上代码中，TccTransactionFilter 类实现了 IFilter 接口，并在 InvokeAsync 方法中拦截 Dubbo 请求。如果当前方法标记了 TccTransactionAttribute，则在 Try 阶段执行请求，并在请求成功后执行 Confirm 阶段。如果请求失败，则执行 Cancel 阶段。如果任何阶段发生异常，则回滚事务。

9. 注册 Dubbo 事务组件 最后，需要在 Dubbo 的配置文件中注册实现的 TCC 事务组件，如下所示：

<dubbo:application name="order-service" />
    <dubbo:registry address="zookeeper://127.0.0.1:2181" />
    <dubbo:protocol name="dubbo" port="20880" />
    <dubbo:provider filter="tccTransactionFilter">
      <dubbo:parameter key="transactionManager" value="DubboTransactionManager" />
      <dubbo:parameter key="transactionRepository" value="InMemoryTccTransactionRepository" />
    </dubbo:provider>

在以上配置中，通过 filter 属性指定了 TccTransactionFilter 过滤器，并通过 parameter 属性指定了实现的事务管理器和事务仓库。

总结

通过以上的实现，可以在 Apache Dubbo .NET 中轻松地实现分布式事务，使用 TCC 模式保证事务的最终一致性。同时，也可以根据具体的业务需求，选择其他的分布式事务解决方案。