如何在.NET 8 Web API中通过Azure PubSub接收纯WebSocket客户端消息

场景概述 #

Python/Java客户端使用原生WebSocket库发送消息，需在.NET 8 Web API中接收消息、处理后存入分布式缓存；客户端不使用Azure PubSub SDK，仅依赖原生WebSocket实现。

疑问解答 #

1. 如何使用ServiceClient从后端服务建立WebSocket连接以接收消息？ #

Azure Web PubSub的ServiceClient不用于建立WebSocket连接接收消息，它的核心作用是服务端管理操作：生成客户端访问令牌、管理组、广播消息等。要接收客户端消息，正确方式是配置Azure Web PubSub的事件推送：

• 在Azure Portal中为目标Hub配置事件处理（如message事件），将事件推送地址指向你的.NET Web API的特定接口（例如/api/webpubsub/receive）
• 客户端通过原生WebSocket连接到Azure Web PubSub后，发送的消息会由Azure服务主动推送到你的API接口，后端直接处理推送请求即可

2. 在Web API中用客户端SDK（托管后台服务）接收消息是否可行？ #

这个方案完全可行，属于服务端以客户端身份加入组监听消息的模式：

• 在.NET Web API中托管一个长期运行的IHostedService后台服务，通过WebPubSubClient连接到Azure Web PubSub并加入目标组，通过GroupMessageReceived事件监听组内消息
• 该模式适合需要实时监听特定组消息的场景，但需注意实现连接断开后的自动重连逻辑，保证稳定性
• 对比事件推送模式：
  • 事件推送是Azure主动将消息推送到API，无需维持长连接，适合全量消息处理
  • 后台监听模式更灵活，适合定向监听特定组消息

代码示例与说明 #

服务端生成客户端连接令牌（供Python/Java客户端使用） #

客户端需先获取带权限的连接URL，由.NET API用ServiceClient生成：

var serviceClient = new WebPubSubServiceClient("<你的连接字符串>", "<Hub名称>");
var clientToken = serviceClient.GetClientAccessToken(new ClientAccessTokenOptions
{
    Groups = new[] { PubSubUtils.GroupName },
    Roles = new[] { "webpubsub.sendToGroup", "webpubsub.joinLeaveGroup" }
});
// 将clientToken.Uri返回给Python/Java客户端，用于建立原生WebSocket连接

Python原生WebSocket发送消息示例 #

客户端遵循Azure Web PubSub的消息格式发送：

import websocket
import json

# 从.NET API获取的带令牌连接URL
ws_uri = "<客户端访问令牌URL>"
ws = websocket.create_connection(ws_uri)

# 构造发送到组的消息
send_msg = json.dumps({
    "type": "sendToGroup",
    "group": "你的组名称",
    "data": "hello_from_python",
    "dataType": "text"
})
ws.send(send_msg)
ws.close()

方案1：Web API事件接收（推荐） #

编写API接口接收Azure推送的消息：

[ApiController]
[Route("api/webpubsub")]
public class WebPubSubEventHandler : ControllerBase
{
    private readonly IDistributedCache _distributedCache;

    public WebPubSubEventHandler(IDistributedCache distributedCache)
    {
        _distributedCache = distributedCache;
    }

    [HttpPost("receive")]
    public async Task<IActionResult> HandleMessage([FromBody] WebPubSubEventRequest request)
    {
        // 可选：验证Azure Web PubSub的请求签名，确保请求合法性
        var messageContent = request.Data.ToString();
        
        // 处理消息并写入分布式缓存
        var cacheKey = $"webpubsub_msg_{Guid.NewGuid()}";
        await _distributedCache.SetStringAsync(cacheKey, messageContent);
        
        return Ok();
    }
}

// 简化的事件请求模型，可根据Azure官方文档补充完整字段
public class WebPubSubEventRequest
{
    public string Type { get; set; }
    public BinaryData Data { get; set; }
    public string Group { get; set; }
}

方案2：后台服务监听组消息 #

实现IHostedService长期监听：

public class GroupMessageListener : IHostedService
{
    private readonly WebPubSubClient _pubSubClient;
    private readonly IDistributedCache _distributedCache;

    public GroupMessageListener(IDistributedCache distributedCache)
    {
        _distributedCache = distributedCache;
        // 生成服务端监听用的访问令牌
        var serviceClient = new WebPubSubServiceClient("<连接字符串>", "<Hub名称>");
        var listenerToken = serviceClient.GetClientAccessToken(new ClientAccessTokenOptions
        {
            Groups = new[] { PubSubUtils.GroupName },
            Roles = new[] { "webpubsub.joinGroup", "webpubsub.receiveGroupMessages" }
        });
        _pubSubClient = new WebPubSubClient(new Uri(listenerToken.Uri));
        _pubSubClient.GroupMessageReceived += OnGroupMessageReceived;
    }

    private async Task OnGroupMessageReceived(GroupMessageReceivedEventArgs args)
    {
        var msgContent = args.Message.Data.ToString();
        var cacheKey = $"group_{PubSubUtils.GroupName}_msg_{Guid.NewGuid()}";
        await _distributedCache.SetStringAsync(cacheKey, msgContent);
    }

    public async Task StartAsync(CancellationToken cancellationToken)
    {
        await _pubSubClient.StartAsync(cancellationToken);
        await _pubSubClient.JoinGroupAsync(PubSubUtils.GroupName, cancellationToken);
    }

    public async Task StopAsync(CancellationToken cancellationToken)
    {
        await _pubSubClient.StopAsync(cancellationToken);
    }
}

// 在Program.cs中注册后台服务
builder.Services.AddHostedService<GroupMessageListener>();

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