如何在.NET Core中从Kafka消费者发JSON并通过WebSocket实时推送到Angular

1 .NET Core 后端改造步骤 #

• 你当前用的WebSocket API本质是SignalR Hub端点，首先需要把Kafka消费者封装为托管后台服务注册到容器中，不要和Hub的生命周期绑定，避免客户端连接断开后消费者就停止运行。
• 后台服务内监听指定Kafka Topic的消息，收到JSON数据后，通过注入的IHubContext<你的Hub类>实例，把数据推送给所有已连接的Angular客户端，也可以根据业务需要推送给指定客户端。

注意：Kafka消费者服务要使用单例生命周期，避免出现重复消费、重复创建消费者实例的问题。

相关代码示例：
Program.cs服务注册与路由配置：

// 注册SignalR服务
builder.Services.AddSignalR();
// 注册Kafka消费者托管服务
builder.Services.AddHostedService<KafkaConsumerHostedService>();

// 配置Hub路由，和前端请求地址对应
app.MapHub<MessageHub>("/kafka-message-hub");

Kafka消费者托管服务示例：

public class KafkaConsumerHostedService : BackgroundService
{
    private readonly IHubContext<MessageHub> _hubContext;
    private IConsumer<string, string> _kafkaConsumer;

    public KafkaConsumerHostedService(IHubContext<MessageHub> hubContext)
    {
        _hubContext = hubContext;
        var consumerConfig = new ConsumerConfig
        {
            BootstrapServers = "你的Kafka服务地址",
            GroupId = "自定义消费组ID",
            AutoOffsetReset = AutoOffsetReset.Earliest
        };
        _kafkaConsumer = new ConsumerBuilder<string, string>(consumerConfig).Build();
        // 订阅需要监听的Kafka Topic
        _kafkaConsumer.Subscribe("目标Topic名称");
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                var consumeResult = _kafkaConsumer.Consume(stoppingToken);
                // 读取Kafka返回的JSON消息
                var messageJson = consumeResult.Message.Value;
                // 推送给所有已连接的SignalR客户端，ReceiveMessage为前端要监听的方法名
                await _hubContext.Clients.All.SendAsync("ReceiveMessage", messageJson, stoppingToken);
            }
            catch (OperationCanceledException)
            {
                _kafkaConsumer.Close();
            }
            catch (Exception ex)
            {
                // 自行添加异常日志处理逻辑
            }
        }
    }
}

SignalR Hub类无需额外逻辑，留空即可：

public class MessageHub : Hub
{
}

2 Angular 前端调整步骤 #

• 你当前使用的@microsoft/signalr包无需更换，正常建立SignalR连接后，监听后端推送的ReceiveMessage方法即可拿到Kafka同步的实时JSON数据。

相关代码示例：
导入包和你当前用法一致：
import * as signalR from "@microsoft/signalr";

连接建立与消息监听逻辑：

private hubConnection: signalR.HubConnection;

ngOnInit(): void {
  // 初始化SignalR连接
  this.hubConnection = new signalR.HubConnectionBuilder()
    .withUrl('http://你的后端接口地址/kafka-message-hub')
    .build();

  // 启动连接
  this.hubConnection.start()
    .then(() => console.log('SignalR连接成功'))
    .catch(err => console.error('连接失败: ', err));

  // 监听后端推送的Kafka消息
  this.hubConnection.on('ReceiveMessage', (kafkaJsonData: any) => {
    console.log('收到Kafka实时数据: ', kafkaJsonData);
    // 此处添加数据接收后的业务处理逻辑
  });
}

• 组件销毁时记得关闭连接，避免资源泄漏：

ngOnDestroy(): void {
  if (this.hubConnection) {
    this.hubConnection.stop();
  }
}

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