构建跨语言微服务架构中基于Tyk与Azure Service Bus的异步事务补偿机制

分布式架构

文章字数: 3.8k

阅读时长: 15 分

一个棘手的现实摆在面前：一个承载核心交易逻辑的Ruby on Rails单体应用，与一个新建的、面向客户的Laravel服务，必须进行数据交互。这个交互的失败率高得无法接受，每一次失败都意味着运营团队需要手动介入，进行数据核对和补偿。问题的根源在于，Rails单体应用处理一个请求的P99耗时超过5秒，且偶发性的数据库锁和内存溢出导致其可用性并非100%。

直接的同步API调用方案，通过API网关（例如Tyk）进行简单的路由转发，第一时间就被否决了。这种方式会将前端Laravel服务的稳定性与脆弱的后端Ruby单体紧紧绑死，形成一个巨大的故障域。任何后端的抖动都会立刻传导至用户侧，造成请求超时和用户体验骤降。这在架构上是不可接受的倒退。

我们需要的是一种具备韧性的、可隔离故障的集成模式。

方案A：同步API编排的诱惑与陷阱

在讨论初期，有人提出了一个看似合理的方案：利用Tyk强大的中间件和插件能力，构建一个同步的API编排层。

其流程大致如下：

sequenceDiagram
    participant C as Client
    participant T as Tyk API Gateway
    participant L as Laravel Service (PHP)
    participant R as Legacy Ruby Service

    C->>+T: POST /api/v2/complex_order
    T->>T: 1. JWT Validation Middleware
    T->>+L: Forward Request
    L->>L: 2. Process initial data, save state as PENDING
    L-->>-T: Respond (e.g., 202 Accepted)
    T-->>-C: Return Order ID (state: PENDING)
    
    par
        T->>+R: 3. Call legacy processing endpoint
        R->>R: Heavy computation (5s+)
        R-->>-T: Return result (SUCCESS/FAILURE)
    and
        T->>+L: 4. (Async) Update order state based on R's response
        L->>L: Update state to COMPLETED or FAILED
        L-->>-T: Acknowledge
    end

这个方案的优点在于，对于调用方来说，逻辑相对简单。Tyk可以配置重试策略，在Ruby服务临时不可用时进行有限次数的重试。

但其致命缺陷在于“同步”二字。即便是Tyk将回调更新状态的步骤伪装成异步，其与Ruby服务的第一次交互仍然是同步阻塞的。如果Ruby服务超时，Tyk的worker进程将被长时间占用。在高并发场景下，这会迅速耗尽网关的连接资源，导致整个API网关的雪崩，影响到其他不相关的服务。此外，如果Ruby服务处理成功但在返回响应时网络中断，Tyk会认为调用失败并发起重试，这破坏了操作的幂等性，可能导致重复处理。

在真实项目中，这种紧耦合的同步调用是分布式系统中的头号杀手。它看似简单，实则将所有服务的可用性风险串联了起来，最终系统的整体SLA（服务等级协议）将是所有串联服务SLA的乘积，一个灾难性的数字。

方案B：基于Azure Service Bus的异步解耦与最终一致性

我们最终选择了基于消息队列的异步通信模式。这引入了“最终一致性”的概念，但换来的是系统的弹性和高可用性。

技术选型上，我们对比了RabbitMQ、Kafka和Azure Service Bus。考虑到团队已有Azure云的使用经验，且Azure Service Bus提供的“会话（Session）”和“死信队列（Dead-letter Queue）”功能非常契合我们的业务场景（需要保证同一订单相关消息的顺序处理，并能自动隔离处理失败的消息），我们最终选择了Azure Service Bus。

新的架构图如下：

graph TD
    subgraph "请求入口"
        Client --> Tyk(Tyk API Gateway)
    end

    subgraph "现代服务 (PHP)"
        Tyk --> Laravel_API(Laravel API)
    end

    subgraph "消息中间件 (Azure)"
        Laravel_API -- 1. Publish OrderCreated Event --> ASB_Topic(Azure Service Bus Topic: orders)
        ASB_Topic -- Route based on subscription rule --> ASB_Queue_Ruby(Queue for Ruby Consumer)
        ASB_Topic -- Route based on subscription rule --> ASB_Queue_Java(Queue for Java Validator)
    end

    subgraph "遗留服务 (Ruby)"
        Ruby_Consumer(Ruby Consumer) -- 2. Process Order --> ASB_Queue_Ruby
        Ruby_Consumer -- 3. Publish OrderProcessed Event --> ASB_Topic
    end
    
    subgraph "新增工具服务 (Java)"
        Jib_Java_Validator(Jib-packaged Java Validator) -- 2a. Validate Data --> ASB_Queue_Java
        Jib_Java_Validator -- 3a. Publish ValidationResult Event --> ASB_Topic
    end

    subgraph "状态同步"
        Laravel_API_Subscriber(Laravel Subscriber) -- 4. Update Final Status --> ASB_Topic
    end

    style Tyk fill:#f9f,stroke:#333,stroke-width:2px
    style ASB_Topic fill:#9cf,stroke:#333,stroke-width:2px

这个架构的核心思想是：

责任单一化：Laravel服务只负责接收请求、校验数据、创建初始订单记录（状态为PENDING），然后立即向Azure Service Bus发布一个OrderCreated事件。它不关心谁来处理、如何处理，完成后迅速响应客户端。
可靠通信：消息总线保证事件的持久化。即使所有下游消费者都宕机，消息也不会丢失。
独立消费：Ruby消费者和新增的Java验证服务独立地订阅它们感兴趣的事件。它们可以按自己的节奏处理，它们的处理失败或成功不会相互影响。
最终一致性：通过订阅下游服务发布的OrderProcessed或ValidationResult等结果事件，Laravel服务最终会更新订单的状态，达到数据的最终一致。

核心实现：代码与配置细节

1. Tyk网关配置：安全与路由

Tyk作为入口，主要负责安全认证和将流量路由到Laravel服务。我们不在这里做任何复杂的业务逻辑编排。

apis/laravel-orders.json Tyk API定义片段：

{
  "name": "Laravel Orders API",
  "api_id": "laravel-orders-api",
  "org_id": "default",
  "use_keyless": false,
  "auth": {
    "auth_header_name": "Authorization"
  },
  "definition": {
    "location": "header",
    "key": "x-api-version"
  },
  "version_data": {
    "not_versioned": true,
    "versions": {
      "Default": {
        "name": "Default",
        "expires": "",
        "paths": {
          "ignored": [],
          "white_list": [],
          "black_list": []
        },
        "use_extended_paths": true,
        "extended_paths": {
            "middleware": [
              {
                "name": "jwt",
                "options": {
                  "validation_strategy": "rs256",
                  "signing_method": "rs256",
                  "source": "ws://my-auth-service",
                  "identity_base_field": "sub",
                  "skip_auth_regex": []
                }
              }
            ]
        }
      }
    }
  },
  "proxy": {
    "listen_path": "/laravel-orders/",
    "target_url": "http://laravel-service.internal:8000/",
    "strip_listen_path": true
  }
}

关注点: 这里我们启用了JWT中间件。所有访问此API的请求都必须携带一个有效的、由内部认证服务签发的JWT。这是保护内部服务的第一道防线。target_url指向Kubernetes集群内部的Laravel服务地址。

2. Laravel事件发布器：保证原子性

在Laravel中，我们必须确保“保存订单到数据库”和“发送消息到Service Bus”这两个操作具备原子性。经典的“发件箱模式”（Outbox Pattern）是解决此问题的标准实践。

app/Services/OrderService.php:

<?php

namespace App\Services;

use App\Models\Order;
use App\Jobs\PublishOrderCreatedEvent;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Log;
use Ramsey\Uuid\Uuid;

class OrderService
{
    /**
     * 创建订单并调度事件发布作业
     *
     * @param array $data
     * @return Order
     * @throws \Throwable
     */
    public function createOrder(array $data): Order
    {
        $order = null;
        
        // 使用数据库事务保证本地状态写入和事件创建的原子性
        DB::transaction(function () use ($data, &$order) {
            $transactionId = Uuid::uuid4()->toString();

            $order = Order::create([
                'user_id' => $data['user_id'],
                'amount' => $data['amount'],
                'status' => 'PENDING',
                'transaction_id' => $transactionId, // 用于下游幂等性控制
            ]);

            // 不直接发送消息，而是将消息内容存入 "outbox" 表
            // 真实的发送操作由一个后台作业异步完成
            // 这样即使消息队列服务暂时不可用，也不会导致数据库回滚
            DB::table('outbox_events')->insert([
                'id' => Uuid::uuid4()->toString(),
                'aggregate_type' => 'Order',
                'aggregate_id' => $order->id,
                'type' => 'OrderCreated',
                'payload' => json_encode([
                    'order_id' => $order->id,
                    'user_id' => $order->user_id,
                    'amount' => $order->amount,
                    'transaction_id' => $transactionId,
                    'timestamp' => now()->toIso8601String(),
                ]),
                'created_at' => now(),
            ]);

        }, 5); // 事务重试5次

        if (!$order) {
            Log::error('Failed to create order after multiple transaction attempts.');
            throw new \RuntimeException('Order creation failed.');
        }

        // 调度一个独立的后台Job来处理outbox表，将事件发布到Azure Service Bus
        // 这个Job可以安全地重试，因为它只负责发送已提交到数据库的事件
        PublishOutboxEvents::dispatch()->onQueue('high-priority');

        return $order;
    }
}

这里的坑在于: 如果你先写数据库，再发送消息，发送消息的步骤可能会失败。如果先发送消息再写库，写库操作可能会失败。这两种情况都会导致数据不一致。Outbox模式通过将“要发送的消息”和业务数据变更放在同一个本地数据库事务中来解决这个问题。一个独立的进程（PublishOutboxEvents Job）负责轮询outbox_events表，并将事件可靠地发送出去，发送成功后再删除或标记为已发送。

3. Ruby消费者：实现幂等性与错误处理

Ruby消费者需要健壮地处理消息，核心是实现幂等性，并妥善处理“毒消息”（Poison Message）。

lib/azure_order_consumer.rb:

require 'azure/service_bus'
require 'redis'
require 'json'
require 'logger'

class AzureOrderConsumer
  SUBSCRIPTION_NAME = 'ruby_legacy_processor'.freeze
  TOPIC_NAME = 'orders'.freeze
  
  def initialize
    # 从环境变量加载配置，这是生产级应用的实践
    connection_string = ENV.fetch('AZURE_SERVICE_BUS_CONNECTION_STRING')
    @service_bus = Azure::ServiceBus::ServiceBusService.new(connection_string)
    @redis = Redis.new(url: ENV.fetch('REDIS_URL'))
    @logger = Logger.new(STDOUT)
  end

  def listen
    @logger.info("Starting to listen on topic '#{TOPIC_NAME}' with subscription '#{SUBSCRIPTION_NAME}'...")
    
    loop do
      # receive_subscription_message 会长轮询，直到有消息或超时
      message = @service_bus.receive_subscription_message(TOPIC_NAME, SUBSCRIPTION_NAME, { timeout: 30 })
      
      next unless message

      begin
        process(message)
        # 显式删除消息，代表处理成功
        @service_bus.delete_subscription_message(message)
        @logger.info("Successfully processed and deleted message: #{message.message_id}")
      rescue => e
        @logger.error("Failed to process message #{message.message_id}. Error: #{e.message}")
        @logger.error(e.backtrace.join("\n"))
        
        # 这里的错误处理至关重要
        # 如果消息的投递次数超过阈值（例如10次），Azure Service Bus会自动将其移入死信队列
        # 我们无需手动处理，只需保证我们的处理逻辑是可重试的
        # 这里我们选择不删除消息，让它在可见性超时后被重新投递
        # 最终它会因为重试次数过多而进入DLQ，等待人工干预
      end
    end
  end

  private

  def process(message)
    payload = JSON.parse(message.body)
    transaction_id = payload['transaction_id']

    # 幂等性检查：使用Redis的SETNX命令，原子地检查并设置锁
    # 锁的key是transaction_id，value是1，过期时间设为较长时间（如1天）以防处理时间过长
    is_new_transaction = @redis.set("idempotency:tx:#{transaction_id}", 1, nx: true, ex: 86400)

    unless is_new_transaction
      @logger.warn("Duplicate message detected, skipping. Transaction ID: #{transaction_id}")
      return # 直接返回，让上层删除消息
    end

    @logger.info("Processing new order: #{payload['order_id']} with transaction ID: #{transaction_id}")
    
    # ... 核心的、可能耗时很长的业务逻辑 ...
    # legacy_order_processing(payload)
    sleep(rand(2..6)) # 模拟耗时操作

    # 处理完成后，发布一个结果事件
    publish_result_event(payload['order_id'], 'PROCESSED', transaction_id)
  end

  def publish_result_event(order_id, status, original_tx_id)
    result_payload = {
      order_id: order_id,
      status: status,
      processed_by: 'ruby_legacy_service',
      original_transaction_id: original_tx_id,
      timestamp: Time.now.utc.iso8601
    }.to_json

    result_message = Azure::ServiceBus::BrokeredMessage.new(result_payload) do |m|
      m.content_type = 'application/json'
      m.label = 'OrderProcessedEvent'
    end
    
    @service_bus.send_topic_message(TOPIC_NAME, result_message)
    @logger.info("Published OrderProcessedEvent for order: #{order_id}")
  end
end

# 启动消费者
AzureOrderConsumer.new.listen

幂等性是关键: 在分布式系统中，消息可能会被重复投递。消费者必须有能力识别并丢弃重复的消息。使用transaction_id配合Redis的原子操作SETNX是实现幂等性的一种常见且高效的方式。
死信队列: 我们没有在代码里写复杂的死信逻辑。这是因为我们利用了Azure Service Bus的内置功能。在创建订阅时，可以配置Max Delivery Count。当一个消息的投递次数超过这个值，它就会被自动转移到与该订阅关联的死信队列（DLQ）中。这极大地简化了消费者的错误处理逻辑。

4. Java验证器与Jib的集成

有时，某些特定任务（如复杂的金融计算、机器学习模型推理）用PHP或Ruby实现性能不佳。这时引入一个用Java或Go编写的高性能微服务是合理的架构决策。这里我们假设需要一个Java服务来做数据校验。

我们选择使用Jib来容器化这个Java应用。

pom.xml (Maven):

<project ...>
    ...
    <dependencies>
        <!-- 例如使用 Spring Boot 和 Spring Cloud Azure Starter -->
        <dependency>
            <groupId>com.azure.spring</groupId>
            <artifactId>spring-cloud-azure-starter-servicebus-jms</artifactId>
        </dependency>
    </dependencies>
    ...
    <build>
        <plugins>
            <plugin>
                <groupId>com.google.cloud.tools</groupId>
                <artifactId>jib-maven-plugin</artifactId>
                <version>3.4.0</version>
                <configuration>
                    <from>
                        <!-- 使用一个精简的、无shell的基础镜像 -->
                        <image>gcr.io/distroless/java17-debian11</image>
                    </from>
                    <to>
                        <image>myregistry/java-validator-service:${project.version}</image>
                        <tags>
                            <tag>latest</tag>
                        </tags>
                    </to>
                    <container>
                        <mainClass>com.example.validator.ValidatorApplication</mainClass>
                        <ports>
                            <port>8080</port>
                        </ports>
                        <jvmFlags>
                            <jvmFlag>-Xms512m</jvmFlag>
                            <jvmFlag>-Xmx1024m</jvmFlag>
                            <!-- G1 GC 是现代JVM的优秀选择 -->
                            <jvmFlag>-XX:+UseG1GC</jvmFlag>
                        </jvmFlags>
                    </container>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

为什么用Jib而不是Dockerfile？
1. 无需Docker守护进程: Jib直接在Maven/Gradle构建流程中生成镜像并推送到仓库。在CI/CD环境中，这意味着你不需要运行一个有特权的Docker-in-Docker服务。
2. 更快的构建: Jib能理解Java项目的结构，它会将项目智能地分层（依赖库、资源文件、class文件）。在代码变更时，通常只有最上层的class文件层需要重新构建，极大地利用了镜像缓存，增量构建速度远快于传统的Dockerfile。
3. 可复现性: Jib构建的镜像默认拥有相同的创建时间戳，确保了只要代码和依赖不变，生成的镜像digest就完全一样，实现了真正的可复现构建。

ValidatorService.java:

import com.azure.spring.messaging.servicebus.core.listener.ServiceBusMessageListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;

@Component
public class OrderValidator {

    private static final Logger LOGGER = LoggerFactory.getLogger(OrderValidator.class);

    // 监听名为 "java_validator" 的订阅
    @JmsListener(destination = "orders", containerFactory = "topicJmsListenerContainerFactory", subscription = "java_validator")
    public void handleMessage(String message) {
        LOGGER.info("Received message: {}", message);
        // ... 解析JSON，执行复杂的校验逻辑 ...
        // 这个服务同样需要实现幂等性消费
        // 校验完成后，也通过ServiceBusTemplate发布结果事件
    }
}

架构的扩展性与局限性

这套基于Tyk网关和Azure Service Bus的异构系统集成架构，为我们带来了期望的韧性和解耦。当Ruby单体应用宕机或处理缓慢时，Laravel应用依然可以正常接收用户请求，用户体验不会受到影响。系统的吞吐量不再受限于最慢的那个组件。未来如果需要引入更多语言编写的服务（例如Python的数据分析服务），只需为orders主题增加一个新的订阅，并编写相应的消费者即可，对现有系统没有任何侵入式修改。

然而，这并非银弹。该架构也带来了新的复杂性。首先，可观测性成为一个巨大的挑战。一个完整的业务流程跨越了PHP、Ruby和Java，并通过消息队列进行中继。传统的单体应用日志和APM工具难以追踪一个请求的全貌。我们必须引入分布式追踪体系（如OpenTelemetry），并确保追踪上下文（Trace Context）能够在HTTP请求和Service Bus消息头中正确传递，这是一个不小的工程挑战。

其次，最终一致性的模型要求业务流程的设计能够容忍数据的短暂不一致。对于某些强一致性要求的场景，例如支付，可能还需要引入Saga模式中的补偿事务逻辑，这会进一步增加业务逻辑的复杂度。当一个多步骤流程中的某一步失败后，需要触发一系列反向操作来回滚已完成的步骤，这对开发和测试都提出了更高的要求。

最后，运维成本有所增加。我们现在需要维护PHP-FPM、Ruby Puma/Unicorn、JVM三种不同的运行时环境，以及Tyk和Azure Service-Bus的配置和监控。这要求团队具备更全面的技术栈知识和更成熟的DevOps文化。

Tyk Azure Service Bus Laravel Ruby Jib

构建基于TiDB的GraphQL服务SLO守护体系：从Code Review到自动化性能卡点

2023-10-27 架构与设计

Code Review GraphQL TiDB SLO Go DevOps

构建基于整洁架构与Redis Streams的幂等事件处理器并使用Ansible与Podman自动化部署

2023-10-27 后端架构

Go DevOps Podman Ansible Clean Architecture Redis Streams