能否将ASP.NET Core WebAPI单独配置为Pub/Sub队列的事件订阅者？

我需要为RabbitMQ、Kafka和Redis三者创建ASP.NET Core C# Pub/Sub监听器的概念验证，目前已经用控制台应用实现了这三种消息队列的发布与订阅功能，但网上大多是控制台示例。我知道WebAPI用HTTP(S)协议，而Pub/Sub常用WS协议，想确认：

能否仅通过WebAPI（不依赖控制台或其他应用）配置为Pub/Sub队列的事件订阅者？如果可以，该如何实现？有没有示例？

附一段可运行的RabbitMQ控制台订阅者代码：

using System.Text;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

Console.WriteLine("Hello, World!");
var factory = new ConnectionFactory { HostName = "localhost" };
using var connection = factory.CreateConnection();
using var channel = connection.CreateModel();

channel.QueueDeclare(queue: "hello",
                     durable: false,
                     exclusive: false,
                     autoDelete: false,
                     arguments: null);

Console.WriteLine(" [*] Waiting for messages.");

var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
    var body = ea.Body.ToArray();
    var message = Encoding.UTF8.GetString(body);
    Console.WriteLine($" [x] Received {message}");
};
channel.BasicConsume(queue: "hello",
                     autoAck: true,
                     consumer: consumer);

Console.WriteLine(" Press [enter] to exit.");
Console.ReadLine();

在ASP.NET Core中，我们可以利用**后台服务（BackgroundService）**托管消息订阅逻辑，让WebAPI启动时自动启动订阅者，无需额外控制台应用。下面分别给出三种消息队列的实现方案：

一、RabbitMQ 实现步骤 #

1. 安装NuGet包 #

Install-Package RabbitMQ.Client

2. 创建后台服务类 #

using System.Text;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using Microsoft.Extensions.Hosting;

public class RabbitMQConsumerService : BackgroundService
{
    private IConnection _connection;
    private IModel _channel;
    private readonly string _queueName = "hello";

    public RabbitMQConsumerService()
    {
        var factory = new ConnectionFactory { HostName = "localhost" };
        _connection = factory.CreateConnection();
        _channel = _connection.CreateModel();

        _channel.QueueDeclare(queue: _queueName,
                             durable: false,
                             exclusive: false,
                             autoDelete: false,
                             arguments: null);
    }

    protected override Task ExecuteAsync(CancellationToken stoppingToken)
    {
        stoppingToken.Register(() =>
        {
            _channel?.Close();
            _connection?.Close();
        });

        var consumer = new EventingBasicConsumer(_channel);
        consumer.Received += (model, ea) =>
        {
            var body = ea.Body.ToArray();
            var message = Encoding.UTF8.GetString(body);
            // 替换为你的业务逻辑，比如写入数据库、调用内部接口等
            Console.WriteLine($"RabbitMQ 收到消息: {message}");
        };

        _channel.BasicConsume(queue: _queueName,
                             autoAck: true,
                             consumer: consumer);

        return Task.CompletedTask;
    }

    public override void Dispose()
    {
        _channel?.Dispose();
        _connection?.Dispose();
        base.Dispose();
    }
}

3. 在Program.cs中注册服务 #

var builder = WebApplication.CreateBuilder(args);

// 注册RabbitMQ消费后台服务
builder.Services.AddHostedService<RabbitMQConsumerService>();

// 常规WebAPI配置
builder.Services.AddControllers();
var app = builder.Build();

app.MapControllers();
app.Run();

二、Kafka 实现步骤 #

1. 安装NuGet包 #

Install-Package Confluent.Kafka

2. 创建后台服务类 #

using Confluent.Kafka;
using Microsoft.Extensions.Hosting;

public class KafkaConsumerService : BackgroundService
{
    private readonly IConsumer<Ignore, string> _consumer;
    private readonly string _topic = "test-topic";

    public KafkaConsumerService()
    {
        var config = new ConsumerConfig
        {
            BootstrapServers = "localhost:9092",
            GroupId = "webapi-consumer-group",
            AutoOffsetReset = AutoOffsetReset.Earliest
        };

        _consumer = new ConsumerBuilder<Ignore, string>(config).Build();
        _consumer.Subscribe(_topic);
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                var consumeResult = _consumer.Consume(stoppingToken);
                // 替换为你的业务逻辑
                Console.WriteLine($"Kafka 收到消息: {consumeResult.Message.Value}");
            }
            catch (ConsumeException e)
            {
                Console.WriteLine($"Kafka消费出错: {e.Error.Reason}");
            }
        }

        _consumer.Close();
    }

    public override void Dispose()
    {
        _consumer.Dispose();
        base.Dispose();
    }
}

3. 在Program.cs中注册服务 #

builder.Services.AddHostedService<KafkaConsumerService>();

三、Redis Pub/Sub 实现步骤 #

1. 安装NuGet包 #

Install-Package StackExchange.Redis

2. 创建后台服务类 #

using StackExchange.Redis;
using Microsoft.Extensions.Hosting;

public class RedisConsumerService : BackgroundService
{
    private readonly IConnectionMultiplexer _redis;
    private readonly string _channelName = "test-channel";

    public RedisConsumerService()
    {
        // 连接本地Redis实例
        _redis = ConnectionMultiplexer.Connect("localhost");
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        var subscriber = _redis.GetSubscriber();
        await subscriber.SubscribeAsync(_channelName, (channel, message) =>
        {
            // 替换为你的业务逻辑
            Console.WriteLine($"Redis 收到消息: {message}");
        });

        // 保持服务运行直到取消信号触发
        await Task.Delay(Timeout.Infinite, stoppingToken);
    }

    public override void Dispose()
    {
        _redis.Dispose();
        base.Dispose();
    }
}

3. 在Program.cs中注册服务 #

builder.Services.AddHostedService<RedisConsumerService>();

关键说明 #

• 后台服务会和WebAPI进程绑定，启动时自动初始化订阅逻辑，停止时自动释放连接资源。
• 消息处理逻辑可根据业务需求自由扩展，比如写入数据库、触发内部业务流程、调用其他微服务等。
• 无需依赖WebSocket，后台服务是独立于HTTP请求的长运行任务，与WebAPI的HTTP服务并行执行。

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