dotnet-realtime-communication

dotnet-realtime-communication

适用于 .NET 应用程序的实时通信模式。比较 SignalR（基于 WebSocket 的全双工通信，并支持自动回退）、Server-Sent Events（SSE，.NET 10 中内置于 ASP.NET Core）、JSON-RPC 2.0（基于任意传输的结构化请求-响应）和 gRPC 流式传输（高性能二进制流式传输）。根据需求提供选择合适协议的指导。

**不在范围内：**HTTP 客户端工厂模式和弹性管道——请参阅 [skill:dotnet-http-client] 和 [skill:dotnet-resilience]。Native AOT 架构和裁剪策略——AOT 编译请参阅 [skill:dotnet-native-aot]，AOT 优先设计模式请参阅 [skill:dotnet-aot-architecture]，安全裁剪开发请参阅 [skill:dotnet-trimming]。Blazor 特定的 SignalR 用法（组件集成、Blazor Server 电路管理、渲染模式交互）——Blazor 托管模型和电路模式请参阅 [skill:dotnet-blazor-patterns]。

交叉引用：gRPC 流式传输实现细节和全部四种流式传输模式请参阅 [skill:dotnet-grpc]。实时通信端点测试请参阅 [skill:dotnet-integration-testing]。Blazor 特定的 SignalR 电路管理和渲染模式交互请参阅 [skill:dotnet-blazor-patterns]。

---

协议对比

协议	方向	传输方式	格式	浏览器支持	最适用场景
SignalR	全双工	WebSocket、SSE、长轮询（自动协商）	JSON 或 MessagePack	支持（JS/TS 客户端）	交互式应用、聊天、仪表板、协作编辑
SSE（.NET 10）	仅服务端到客户端	HTTP/1.1+	文本（通常为 JSON 行）	支持（原生 EventSource API）	通知、实时动态、状态更新
JSON-RPC 2.0	请求-响应	任意（HTTP、WebSocket、stdio）	JSON	取决于传输方式	工具协议（LSP）、基于简单传输的结构化 RPC
gRPC 流式传输	全部四种模式	HTTP/2	Protobuf（二进制）	有限（gRPC-Web）	服务间通信、高吞吐量、低延迟流式传输

如何选择

• SignalR：需要与浏览器客户端进行双向实时通信。SignalR 会自动协商传输方式（优先使用 WebSocket，不可用时回退到 SSE，然后回退到长轮询）。适用于客户端需要实时收发消息的场景。
• SSE（.NET 10 内置）：只需要服务端向客户端推送。无需双向通信时，它比 SignalR 更简单。内置于 .NET 10 的 ASP.NET Core 中，无需额外软件包。可与浏览器的原生 EventSource API 配合使用。
• JSON-RPC 2.0：需要通过简单传输方式实现结构化请求-响应语义。Language Server Protocol（LSP）和一些 .NET 工具会使用它。它不是流式传输协议——当你需要通过 WebSocket 或 stdio 使用带类型参数的命名方法时，可选用它。
• gRPC 流式传输：追求最高性能的服务间流式传输。支持全部四种流式传输模式（单项、服务端流式、客户端流式、双向流式）。适用于两个端点都是 .NET 服务或兼容 gRPC 的场景。实现细节请参阅 [skill:dotnet-grpc]。

---

SignalR

SignalR 提供实时 Web 功能，并支持自动连接管理和传输方式协商。

服务端设置

var builder = WebApplication.CreateBuilder(args);

builder.Services.AddSignalR(options =>
{
    options.EnableDetailedErrors = builder.Environment.IsDevelopment();
    options.MaximumReceiveMessageSize = 64 * 1024; // 64 KB
    options.KeepAliveInterval = TimeSpan.FromSeconds(15);
});

var app = builder.Build();

app.MapHub<NotificationHub>("/hubs/notifications");

Hub 实现

public sealed class NotificationHub(
    ILogger<NotificationHub> logger) : Hub
{
    public override async Task OnConnectedAsync()
    {
        var userId = Context.UserIdentifier;
        if (userId is not null)
        {
            await Groups.AddToGroupAsync(Context.ConnectionId, $"user:{userId}");
        }

        await base.OnConnectedAsync();
    }

    // Client-to-server method
    public async Task SendMessage(string channel, string message)
    {
        // Broadcast to all clients in the channel group
        await Clients.Group(channel).SendAsync("ReceiveMessage",
            Context.UserIdentifier, message);
    }

    // Server-to-client streaming
    public async IAsyncEnumerable<StockPrice> StreamPrices(
        string symbol,
        [EnumeratorCancellation] CancellationToken cancellationToken)
    {
        while (!cancellationToken.IsCancellationRequested)
        {
            yield return await GetLatestPrice(symbol, cancellationToken);
            await Task.Delay(1000, cancellationToken);
        }
    }
}

强类型 Hub

使用接口确保客户端方法调用在编译时安全：

public interface INotificationClient
{
    Task ReceiveMessage(string user, string message);
    Task OrderStatusChanged(int orderId, string status);
}

public sealed class NotificationHub(
    ILogger<NotificationHub> logger) : Hub<INotificationClient>
{
    public async Task SendMessage(string channel, string message)
    {
        // Compile-time checked -- no magic strings
        await Clients.Group(channel).ReceiveMessage(
            Context.UserIdentifier!, message);
    }
}

在 Hub 外部发送消息

注入 IHubContext，以便从后台服务或控制器发送消息：

public sealed class OrderService(
    IHubContext<NotificationHub, INotificationClient> hubContext)
{
    public async Task UpdateOrderStatus(int orderId, string userId, string status)
    {
        // Send to specific user group
        await hubContext.Clients.Group($"user:{userId}")
            .OrderStatusChanged(orderId, status);
    }
}

传输方式协商

SignalR 会自动协商最佳传输方式：

1. WebSocket（首选）——全双工，延迟最低
2. Server-Sent Events——仅服务端到客户端；WebSocket 不可用时回退到此方式
3. 长轮询——通用回退方式，延迟最高

需要时可强制指定传输方式：

// Server: disable specific transports
app.MapHub<NotificationHub>("/hubs/notifications", options =>
{
    options.Transports = HttpTransportType.WebSockets |
                         HttpTransportType.ServerSentEvents;
    // Disables Long Polling
});

MessagePack 协议

使用 MessagePack 缩小负载并加快序列化：

// Server
builder.Services.AddSignalR()
    .AddMessagePackProtocol();

// Client (JavaScript)
// new signalR.HubConnectionBuilder()
//     .withUrl("/hubs/notifications")
//     .withHubProtocol(new signalR.protocols.msgpack.MessagePackHubProtocol())
//     .build();

连接生命周期

重写 OnConnectedAsync 和 OnDisconnectedAsync 以管理连接状态：

public sealed class NotificationHub(
    ILogger<NotificationHub> logger,
    IConnectionTracker tracker) : Hub<INotificationClient>
{
    public override async Task OnConnectedAsync()
    {
        var userId = Context.UserIdentifier;
        var connectionId = Context.ConnectionId;

        logger.LogInformation("Client {ConnectionId} connected (user: {UserId})",
            connectionId, userId);

        // Track connection for presence features
        if (userId is not null)
        {
            await tracker.AddConnectionAsync(userId, connectionId);
            await Groups.AddToGroupAsync(connectionId, $"user:{userId}");
        }

        await base.OnConnectedAsync();
    }

    public override async Task OnDisconnectedAsync(Exception? exception)
    {
        var userId = Context.UserIdentifier;
        var connectionId = Context.ConnectionId;

        if (exception is not null)
        {
            logger.LogWarning(exception,
                "Client {ConnectionId} disconnected with error", connectionId);
        }

        if (userId is not null)
        {
            await tracker.RemoveConnectionAsync(userId, connectionId);
        }

        await base.OnDisconnectedAsync(exception);
    }
}

组管理

组提供轻量级的发布/订阅机制。一个连接可以加入多个组，组成员关系按连接管理：

public sealed class ChatHub : Hub<IChatClient>
{
    // Join a room (called by clients)
    public async Task JoinRoom(string roomName)
    {
        await Groups.AddToGroupAsync(Context.ConnectionId, roomName);
        await Clients.Group(roomName).UserJoined(Context.UserIdentifier!, roomName);
    }

    // Leave a room
    public async Task LeaveRoom(string roomName)
    {
        await Groups.RemoveFromGroupAsync(Context.ConnectionId, roomName);
        await Clients.Group(roomName).UserLeft(Context.UserIdentifier!, roomName);
    }

    // Send to specific group
    public async Task SendToRoom(string roomName, string message)
    {
        await Clients.Group(roomName).ReceiveMessage(
            Context.UserIdentifier!, message);
    }

    // Send to all except caller
    public async Task BroadcastExceptSelf(string message)
    {
        await Clients.Others.ReceiveMessage(
            Context.UserIdentifier!, message);
    }
}

组不会持久化——连接断开时会清除组成员关系。如有需要（例如从数据库或缓存中读取），请在 OnConnectedAsync 中将连接重新加入组。

客户端到服务端流式传输

客户端可以使用 IAsyncEnumerable<T> 或 ChannelReader<T> 向中心传输数据：

public sealed class UploadHub : Hub
{
    // Accept a stream of items from the client
    public async Task UploadData(
        IAsyncEnumerable<SensorReading> stream,
        CancellationToken cancellationToken)
    {
        await foreach (var reading in stream.WithCancellation(cancellationToken))
        {
            await ProcessReading(reading);
        }
    }
}

身份验证

SignalR 使用与 ASP.NET Core 宿主相同的身份验证方式。对于 WebSocket 连接，由于 WebSocket 不支持自定义标头，因此访问令牌通过查询字符串传递：

// Server: configure JWT for SignalR
builder.Services.AddAuthentication(JwtBearerDefaults.AuthenticationScheme)
    .AddJwtBearer(options =>
    {
        options.Authority = "https://identity.example.com";
        options.Events = new JwtBearerEvents
        {
            OnMessageReceived = context =>
            {
                // Read token from query string for WebSocket requests
                var accessToken = context.Request.Query["access_token"];
                var path = context.HttpContext.Request.Path;
                if (!string.IsNullOrEmpty(accessToken) &&
                    path.StartsWithSegments("/hubs"))
                {
                    context.Token = accessToken;
                }
                return Task.CompletedTask;
            }
        };
    });

builder.Services.AddAuthorization();

var app = builder.Build();

app.UseAuthentication();
app.UseAuthorization();

app.MapHub<NotificationHub>("/hubs/notifications")
    .RequireAuthorization();

在中心中访问 Context.UserIdentifier，以识别经过身份验证的用户。默认情况下，该属性映射到 ClaimTypes.NameIdentifier 声明。可使用 IUserIdProvider 自定义：

public sealed class EmailUserIdProvider : IUserIdProvider
{
    public string? GetUserId(HubConnectionContext connection)
    {
        return connection.User?.FindFirst(ClaimTypes.Email)?.Value;
    }
}

// Register
builder.Services.AddSingleton<IUserIdProvider, EmailUserIdProvider>();

使用 Backplane 扩展

对于多服务器部署，使用 backplane 在各实例之间同步消息。没有 backplane 时，在一台服务器上发送的消息对连接到其他服务器的客户端不可见。

Redis backplane：

builder.Services.AddSignalR()
    .AddStackExchangeRedis(builder.Configuration.GetConnectionString("Redis")!,
        options =>
        {
            options.Configuration.ChannelPrefix =
                RedisChannel.Literal("MyApp:");
        });

Azure SignalR Service（托管 backplane）：

builder.Services.AddSignalR()
    .AddAzureSignalR(builder.Configuration["Azure:SignalR:ConnectionString"]);

Azure SignalR Service 会完全卸载连接管理——ASP.NET Core 服务器负责处理中心逻辑，而 Azure 管理 WebSocket 连接、扩缩容和消息路由。

---

Server-Sent Events（SSE）——.NET 10

.NET 10 为 ASP.NET Core 增加了内置 SSE 支持，无需额外包即可轻松实现从服务端到客户端的流式传输。

Minimal API 终结点

app.MapGet("/events/orders", async (
    OrderEventService eventService,
    CancellationToken cancellationToken) =>
{
    // TypedResults.ServerSentEvents returns an SSE response
    return TypedResults.ServerSentEvents(
        eventService.GetOrderEventsAsync(cancellationToken));
});

事件源实现

public sealed class OrderEventService
{
    public async IAsyncEnumerable<SseItem<OrderEvent>> GetOrderEventsAsync(
        [EnumeratorCancellation] CancellationToken cancellationToken)
    {
        while (!cancellationToken.IsCancellationRequested)
        {
            var evt = await WaitForNextEvent(cancellationToken);
            yield return new SseItem<OrderEvent>(evt, "order-update");
        }
    }
}

浏览器客户端

const source = new EventSource('/events/orders');

source.addEventListener('order-update', (event) => {
    const order = JSON.parse(event.data);
    updateDashboard(order);
});

source.onerror = () => {
    // EventSource automatically reconnects
    console.log('SSE connection lost, reconnecting...');
};

何时选择 SSE 而非 SignalR

• 仅需单向推送——不需要客户端到服务端消息时，SSE 更简单
• 浏览器原生支持——无需 JavaScript 库（使用 EventSource API）
• 自动重新连接——浏览器会使用 Last-Event-ID 自动重新连接
• 兼容 HTTP/1.1——可通过不支持 WebSocket 升级的代理工作

---

JSON-RPC 2.0

JSON-RPC 2.0 是一种无状态、与传输方式无关的远程过程调用协议，采用 JSON 编码。它是语言服务器协议（LSP）的基础，也用于一些 .NET 工具场景。

协议结构

// Request
{"jsonrpc": "2.0", "method": "textDocument/completion", "params": {...}, "id": 1}

// Response
{"jsonrpc": "2.0", "result": {...}, "id": 1}

// Notification (no response expected)
{"jsonrpc": "2.0", "method": "textDocument/didChange", "params": {...}}

// Error
{"jsonrpc": "2.0", "error": {"code": -32600, "message": "Invalid Request"}, "id": 1}

StreamJsonRpc（.NET 库）

StreamJsonRpc 是用于 JSON-RPC 2.0 的主要 .NET 库：

<PackageReference Include="StreamJsonRpc" Version="2.*" />

// Server: expose methods via JSON-RPC over a stream
using StreamJsonRpc;

public sealed class CalculatorService
{
    public int Add(int a, int b) => a + b;
    public Task<double> DivideAsync(double a, double b) =>
        b == 0 ? throw new ArgumentException("Division by zero")
               : Task.FromResult(a / b);
}

// Wire up over a WebSocket -- UseWebSockets() is required for upgrade handling
app.UseWebSockets();
app.Map("/jsonrpc", async (HttpContext context) =>
{
    if (!context.WebSockets.IsWebSocketRequest)
    {
        context.Response.StatusCode = 400;
        return;
    }

    var ws = await context.WebSockets.AcceptWebSocketAsync();
    using var rpc = new JsonRpc(new WebSocketMessageHandler(ws));
    rpc.AddLocalRpcTarget(new CalculatorService());
    rpc.StartListening();
    await rpc.Completion;
});

// Client
using var ws = new ClientWebSocket();
await ws.ConnectAsync(new Uri("ws://localhost:5000/jsonrpc"),
    CancellationToken.None);

using var rpc = new JsonRpc(new WebSocketMessageHandler(ws));
rpc.StartListening();

var result = await rpc.InvokeAsync<int>("Add", 2, 3);
// result == 5

何时使用 JSON-RPC 2.0

• 构建或集成语言服务器协议（LSP）实现
• 在 gRPC 过于笨重时，通过 WebSocket 或 stdio 实现简单 RPC
• 与使用 JSON-RPC 的非 .NET 系统互操作
• 工具和编辑器集成

---

gRPC 流式传输

有关完整的 gRPC 实现详情，请参阅 [skill:dotnet-grpc]，其中涵盖所有四种流式传输模式（一元、服务端流式传输、客户端流式传输、双向流式传输）、身份验证、负载均衡和运行状况检查。

快速决策：gRPC 流式传输、SignalR 还是 SSE

需求	选择
服务到服务，双方均使用 .NET	gRPC 流式传输
浏览器客户端需要双向通信	SignalR
浏览器客户端仅需要服务端推送	SSE
最高吞吐量、二进制负载	gRPC 流式传输
浏览器客户端自动重新连接	SSE（原生）或 SignalR（内置）
多种客户端平台（JS、移动端、.NET）	SignalR

---

核心原则

• 面向浏览器的实时通信默认使用 SignalR —— 它开箱即用地处理传输协商、重新连接和分组
• 简单的服务器推送使用 SSE —— .NET 10 内置支持使其成为单向通知中最轻量的选项
• 服务间通信使用 gRPC 流式传输 —— 性能最高、契约强类型，并支持全部四种流式模式
• 工具协议使用 JSON-RPC 2.0 —— 需要通过简单传输方式（WebSocket、stdio）实现结构化 RPC 时使用
• 使用强类型 Hub —— Hub<T> 可在编译时而非运行时捕获方法名称拼写错误
• 使用 backplane 扩展 SignalR —— 多服务器部署时使用 Redis 或 Azure SignalR Service

有关 AOT 编译管道，请参阅 [skill:dotnet-native-aot]；有关兼容 AOT 的实时通信模式，请参阅 [skill:dotnet-aot-architecture]。

---

Agent 常见问题

1. SSE 足够时不要使用 SignalR —— 如果只需要服务器向客户端推送，而不需要双向通信，SSE 更简单、也更轻量。
2. 客户端使用 MessagePack 时，服务器不要忘记调用 AddMessagePackProtocol() —— 协议不匹配会导致连接静默失败。
3. 除非确有需要，否则不要将 Long Polling 传输与 SignalR 搭配使用 —— 与 WebSocket 相比，它的延迟和服务器资源占用都显著更高。
4. 不要长期存储连接 ID —— SignalR 连接 ID 会在重新连接时更改。使用用户标识符或组进行寻址。
5. 不要直接向浏览器使用 gRPC 流式传输 —— 浏览器原生不支持 HTTP/2 trailers。请通过代理使用 gRPC-Web，或改用 SignalR/SSE。
6. 不要混淆 SSE 和 WebSocket —— SSE 是单向的（仅服务器到客户端）。如果需要客户端到服务器的消息，请使用 SignalR 或直接使用 WebSocket。
7. 使用 SignalR 和 JWT 时不要忘记 OnMessageReceived —— WebSocket 连接在初始握手后无法发送自定义 HTTP 标头。必须在 JwtBearerEvents.OnMessageReceived 中从查询字符串读取访问令牌。
8. 不要假设重新连接后组成员身份仍然有效 —— 组与连接 ID 绑定，而连接 ID 会在重新连接时更改。请在 OnConnectedAsync 中将连接重新加入组。
9. 部署多服务器 SignalR 时不要省略 backplane —— 如果没有 Redis 或 Azure SignalR Service，在一个服务器实例上发送的消息对其他实例上的连接不可见。

---

版权归属

改编自 Aaronontheweb/dotnet-skills（MIT 许可证）。

---

参考资料

• SignalR 概述
• SignalR Hub
• .NET 10 中的服务器发送事件
• StreamJsonRpc
• gRPC 流式传输
• 使用 Redis 扩展 SignalR
• SignalR 身份验证和授权
• Azure SignalR Service