Spring WebFlux 如何等待并关联 webhook 回调？

2026-10-24支付阿杰

结论

不要阻塞线程等待 webhook，也不要调用 block()。可以按下面的方式处理：

为每次支付生成唯一的关联 ID。
发送终端激活请求前，先注册一个 Sinks.One<PaymentResult>。
将关联 ID 随请求传给 payment gateway。
webhook 到达后，根据关联 ID 找到对应的 Sinks.One 并完成它。
GraphQL subscription 返回该 Sink 对应的 Mono 或 Flux，由 WebFlux 异步等待结果。

如果一次支付只有一个最终结果，适合使用 Mono 和 Sinks.One。如果需要依次推送“终端已激活、等待付款、支付成功”等状态，则使用 Flux 和 Sinks.Many。

为什么不能直接“等待”

webhook 和终端激活请求是两个独立的 HTTP 调用，甚至可能由不同的服务器实例处理。WebFlux 无法自动判断某个 webhook 对应哪一次请求。要解决的不是选择哪种等待 API，而是为两个请求建立可靠的关联关系。

典型流程如下：

GraphQL Subscription
        │
        ├─ 生成 correlationId
        ├─ 注册 Sinks.One
        └─ 激活支付终端，并发送 correlationId
                          │
                          ▼
                    Payment Gateway
                          │
                          ▼
Webhook ── correlationId ── 找到 Sinks.One ── 推送支付结果

这里的“等待”不会占用线程。订阅会保持未完成，直到 Sink 收到结果、操作超时或客户端断开连接。

单实例实现示例

下面的实现适用于单实例应用，也可以作为完整方案的基础。

支付等待注册表

import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

@Component
public class PaymentWaitRegistry {

    private final ConcurrentMap<String, PendingPayment> pendingPayments =
            new ConcurrentHashMap<>();

    public PendingPayment register(String correlationId) {
        PendingPayment pending = new PendingPayment(
                Sinks.one()
        );

        PendingPayment existing =
                pendingPayments.putIfAbsent(correlationId, pending);

        if (existing != null) {
            throw new IllegalStateException(
                    "Duplicate correlationId: " + correlationId
            );
        }

        return pending;
    }

    public boolean complete(
            String correlationId,
            PaymentResult result
    ) {
        PendingPayment pending = pendingPayments.get(correlationId);

        if (pending == null) {
            return false;
        }

        Sinks.EmitResult emitResult =
                pending.sink().tryEmitValue(result);

        return emitResult.isSuccess();
    }

    public void remove(
            String correlationId,
            PendingPayment pending
    ) {
        pendingPayments.remove(correlationId, pending);
    }

    public record PendingPayment(
            Sinks.One<PaymentResult> sink
    ) {
        public Mono<PaymentResult> asMono() {
            return sink.asMono();
        }
    }
}

这里使用 ConcurrentHashMap，因为 GraphQL subscription 和 webhook 通常由不同的线程处理。

webhook 场景更适合使用 tryEmitValue()，而不是 emitValue()。重复 webhook、超时后到达的回调或客户端取消订阅，都可能导致代码再次尝试完成 Sink。使用 tryEmitValue() 可以明确处理这些情况。

激活终端并等待付款

import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.util.UUID;

@Service
public class PaymentService {

    private final PaymentGatewayClient gatewayClient;
    private final PaymentWaitRegistry waitRegistry;

    public PaymentService(
            PaymentGatewayClient gatewayClient,
            PaymentWaitRegistry waitRegistry
    ) {
        this.gatewayClient = gatewayClient;
        this.waitRegistry = waitRegistry;
    }

    public Mono<PaymentResult> activateAndAwaitPayment(
            ActivatePaymentCommand command
    ) {
        return Mono.defer(() -> {
            String correlationId =
                    UUID.randomUUID().toString();

            PaymentWaitRegistry.PendingPayment pending =
                    waitRegistry.register(correlationId);

            return gatewayClient
                    .activateTerminal(command, correlationId)
                    .then(pending.asMono())
                    .timeout(Duration.ofMinutes(5))
                    .doFinally(signalType ->
                            waitRegistry.remove(
                                    correlationId,
                                    pending
                            )
                    );
        });
    }
}

必须先注册 Sink，再激活终端。如果 payment gateway 响应很快，webhook 可能在应用建立等待记录之前到达。

即使 Sinks.One 在订阅者开始监听前收到结果，它也会保存该结果，并在订阅建立后发送出去。因此，即便 webhook 在激活请求执行期间提前到达，也不会丢失结果。

Mono.defer() 可以保证只有 GraphQL 客户端真正订阅时，系统才会创建关联 ID 和等待记录。

Payment gateway 客户端

具体字段由 payment gateway 的 API 决定。应用需要把自己生成的关联 ID 传给对方，并要求对方在 webhook 中原样返回。

import reactor.core.publisher.Mono;

public interface PaymentGatewayClient {

    Mono<Void> activateTerminal(
            ActivatePaymentCommand command,
            String correlationId
    );
}

例如，请求可以使用类似字段：

{
  "amount": 1000,
  "currency": "CNY",
  "merchantReference": "b849c04e-72c0-4e91-92d1-fb0bcdcb1739"
}

对应的 webhook 应包含相同的值：

{
  "merchantReference": "b849c04e-72c0-4e91-92d1-fb0bcdcb1739",
  "status": "PAID",
  "transactionId": "txn-123456"
}

字段名称应以 payment gateway 实际支持的协议为准。如果网关不允许传入自定义关联 ID，可以改用激活响应中返回的 payment ID。不过，这会产生注册记录与 webhook 并发到达的竞态问题。此时应先持久化 payment ID 与业务订单的映射，再处理 webhook。

webhook 处理

webhook Controller 不需要等待 GraphQL subscription 完成。它只负责验证回调、记录支付结果、完成对应的 Sink，并尽快返回响应。

import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Mono;

@RestController
@RequestMapping("/webhooks/payments")
public class PaymentWebhookController {

    private final PaymentWaitRegistry waitRegistry;
    private final PaymentWebhookVerifier webhookVerifier;

    public PaymentWebhookController(
            PaymentWaitRegistry waitRegistry,
            PaymentWebhookVerifier webhookVerifier
    ) {
        this.waitRegistry = waitRegistry;
        this.webhookVerifier = webhookVerifier;
    }

    @PostMapping
    public Mono<ResponseEntity<Void>> handleWebhook(
            @RequestBody PaymentWebhookPayload payload,
            @RequestHeader("X-Signature") String signature
    ) {
        return webhookVerifier
                .verify(payload, signature)
                .then(Mono.fromSupplier(() -> {
                    PaymentResult result = new PaymentResult(
                            payload.transactionId(),
                            payload.status()
                    );

                    waitRegistry.complete(
                            payload.merchantReference(),
                            result
                    );

                    return ResponseEntity.ok().<Void>build();
                }));
    }
}

实际项目中的签名校验可能需要使用原始请求体，而不能使用反序列化后的对象。具体方式应以网关的签名协议为准。

重复 webhook 通常不应视为服务器错误。支付平台没有及时收到成功响应时可能会重试，因此支付结果处理必须具备幂等性。例如，可以根据 transactionId 或 payment ID 防止重复入账。

GraphQL subscription

如果 subscription 只返回最终支付结果，可以将 Mono 转成 Flux：

import org.springframework.graphql.data.method.annotation.Argument;
import org.springframework.graphql.data.method.annotation.SubscriptionMapping;
import org.springframework.stereotype.Controller;
import reactor.core.publisher.Flux;

@Controller
public class PaymentSubscriptionController {

    private final PaymentService paymentService;

    public PaymentSubscriptionController(
            PaymentService paymentService
    ) {
        this.paymentService = paymentService;
    }

    @SubscriptionMapping
    public Flux<PaymentResult> waitForPayment(
            @Argument String terminalId,
            @Argument long amount
    ) {
        ActivatePaymentCommand command =
                new ActivatePaymentCommand(terminalId, amount);

        return paymentService
                .activateAndAwaitPayment(command)
                .flux();
    }
}

客户完成支付后，webhook 会完成 Sinks.One，支付结果随即通过 GraphQL WebSocket 推送给客户端。如果操作超时、终端激活失败或客户端取消订阅，doFinally 会清理注册表中的记录。

多实例部署不能只使用内存 Map

ConcurrentHashMap 方案只适用于单实例，或者能够保证 GraphQL subscription 和 webhook 始终路由到同一实例的环境。多实例部署时可能出现以下情况：

subscription 在实例 A 上建立。
webhook 被负载均衡器转发到实例 B。
实例 B 的内存中没有对应的 Sink。

生产环境通常需要持久化支付状态，并通过 Redis Pub/Sub、Kafka、RabbitMQ 等消息设施，将 webhook 事件发送给正在等待该支付结果的实例。消息的 key 或 topic routing key 应使用 correlationId。

更可靠的处理顺序通常是：

webhook 验证签名。
根据 payment ID 或 correlation ID 幂等地更新数据库。
发布 PaymentCompleted 事件。
subscription 所在的实例收到事件，并完成本地 Sink。
webhook 返回成功响应。

数据库是支付状态的事实来源，消息只负责实时通知。这样即使服务重启、消息重复、客户端断线或 webhook 提前到达，客户端仍然可以重新查询支付状态。

需要特别注意的事项

不要使用 Thread.sleep()、CountDownLatch.await() 或 Mono.block()，这些调用会占用或阻塞 WebFlux 线程。
关联 ID 必须唯一，最好同时将它用作幂等键。
发送激活请求前先注册等待对象，防止过早到达的 webhook 丢失。
必须设置超时，否则未完成的支付会长期占用内存。
客户端断开 GraphQL WebSocket 后，应清理对应的等待记录。
webhook 必须验证签名、时间戳或其他真实性凭据。
支付结果必须先可靠写入数据库，再通知 GraphQL 客户端。
不要假设 webhook 只会调用一次，也不要假设它会按顺序到达。
Sinks.One 只能处理单个 JVM 内部的异步衔接，不能代替分布式状态存储。
在支付业务中，即使客户端收到了实时通知，仍应以服务端保存的最终支付状态为准。