如何在不影响APIM SSE流式传输的前提下异步记录响应至Event Hub？

核心思路 #

利用APIM的异步处理能力和响应流式转发特性，在不阻塞主响应流的前提下，将SSE分片数据异步投递到Event Hub或Service Bus，同时保证客户端能实时接收流式数据。

具体实现步骤 #

1. 使用<send-request>策略异步转发分片数据 #

在APIM的出站策略中，不要一次性读取完整响应体，而是通过<set-body>结合@(context.Response.Body.AsStream())保持流式传输，同时并行启动异步请求将分片数据发送到Event Hub：

<outbound>
    <!-- 保持原响应流式传输到客户端 -->
    <set-body>@(context.Response.Body.AsStream())</set-body>
    
    <!-- 异步发送SSE分片到Event Hub -->
    <send-request mode="new" response-variable-name="ignoreResponse" timeout="20">
        <set-url>https://<your-eventhub-namespace>.servicebus.windows.net/<your-eventhub-name>/messages</set-url>
        <set-method>POST</set-method>
        <set-header name="Authorization" exists-action="override">
            <value>@(SharedAccessSignatureBuilder.BuildSharedAccessSignature("RootManageSharedAccessKey", "<your-sas-key>", "<your-eventhub-namespace>.servicebus.windows.net", TimeSpan.FromMinutes(5)))</value>
        </set-header>
        <set-header name="Content-Type" exists-action="override">
            <value>application/json</value>
        </set-header>
        <set-body>@{
            // 读取当前SSE分片内容，根据实际响应格式调整处理逻辑
            string sseChunk = context.Response.Body.As<string>(true);
            // 附加客户端/用户标识等元数据，用于后续Token消耗计算
            return JsonConvert.SerializeObject(new {
                UserId = context.Request.Headers.GetValueOrDefault("X-User-Id", "unknown"),
                ClientId = context.Request.Headers.GetValueOrDefault("X-Client-Id", "unknown"),
                SseContent = sseChunk,
                Timestamp = DateTime.UtcNow
            });
        }</set-body>
    </send-request>
</outbound>

说明：mode="new"会创建独立的异步请求，不会阻塞主响应流；response-variable-name="ignoreResponse"表示无需等待异步请求的响应结果，避免拖慢主流程。

2. 优化SSE分片处理逻辑 #

如果存在非流式请求，可结合<choose>策略判断响应格式，仅对SSE类型的响应触发异步记录：

<outbound>
    <choose>
        <when condition="@(context.Response.Headers.GetValueOrDefault("Content-Type", "").Contains("text/event-stream"))">
            <!-- 保持流式传输到客户端 -->
            <set-body>@(context.Response.Body.AsStream())</set-body>
            
            <!-- 异步发送每个SSE分片 -->
            <send-request mode="new" response-variable-name="ignoreResponse" timeout="20">
                <!-- 同上述Event Hub请求配置 -->
            </send-request>
        </when>
        <otherwise>
            <!-- 非流式响应的常规处理逻辑 -->
        </otherwise>
    </choose>
</outbound>

3. Service Bus适配调整 #

如果选择Service Bus作为存储目标，只需修改<send-request>中的URL和授权逻辑：

• URL格式：https://<your-servicebus-namespace>.servicebus.windows.net/<your-queue-name>/messages
• 授权方式同样使用SAS签名，逻辑与Event Hub一致。

关键注意事项 #

• 避免重复读取响应体：使用AsStream()或As<string>(true)（true表示保留原始流），防止主响应流被消耗导致客户端无法接收数据。
• 异步请求超时设置：合理设置timeout参数，避免异步请求长时间占用APIM资源。
• 元数据完整性：务必携带用户ID、客户端ID等标识，后续才能准确计算不同主体的Token消耗。
• 监控与排查：定期监控APIM异步请求成功率、Event Hub/Service Bus消息投递情况，确保数据不丢失。

内容的提问来源于stack exchange，提问作者Jayendran