HMENetCore.RabbitMQ 6.0.47

HMENetCore.RabbitMQ

简介

HMENetCore.RabbitMQ 基于 RabbitMQ .NET 客户端的封装库，提供了更简单易用的 API 来处理 RabbitMQ 的连接、生产者和消费者操作。

功能特性

• 自动连接管理，支持断线重连
• 多主机支持，自动故障转移
• 线程安全的连接和通道管理
• 支持多种消息模式：普通模式/Work模式、发布/订阅模式、路由模式、Topic模式
• 消息确认和拒绝机制
• 消息限流控制

.NET支持

• 跨平台支持: Supports .NET Standard 2.1, .NET 6, .NET 8, and .NET 9.

安装

dotnet add package HMENetCore.RabbitMQ --version 6.0.47

配置连接

添加配置到 appsettings.json：

{
  "RabbitMQConfig": {
    "Hosts": [ "rabbitmq1", "rabbitmq2", "rabbitmq3" ],
    "Port": 5672,
    "UserName": "guest",
    "Password": "guest",
    "VirtualHost": "/"
  }
}

服务注册

var rabbitMQConfig = Configuration.GetSection("RabbitMQConfig").Get<RabbitMQConfig>();
builder.Services.AddRabbitMQSetup(rabbitMQConfig);

生产者示例

public class MessageService
{
    private readonly RabbitMQProducer _producer;
    
    public MessageService(RabbitMQProducer producer)
    {
        _producer = producer;
    }
    
    public async Task SendMessageAsync(string queueName, string message)
    {
        await _producer.PublishAsync(queueName, message, options =>
        {
            options.Durable = true; // 持久化队列
        });
    }
    
    public async Task SendToExchangeAsync(string exchange, string routingKey, string message)
    {
        await _producer.PublishAsync(exchange, routingKey, message, options =>
        {
            options.Type = RabbitMQExchangeType.Direct;
            options.Durable = true;
        });
    }
}

消费者示例

public class MessageProcessor : IHostedService, IDisposable
{
    private readonly RabbitMQConsumer _consumer;
    private readonly ILogger<MessageProcessor> _logger;
    private ListenResult _listenResult;
    
    public MessageProcessor(RabbitMQConsumer consumer, ILogger<MessageProcessor> logger)
    {
        _consumer = consumer;
        _logger = logger;
    }
    
    public async Task StartAsync(CancellationToken cancellationToken)
    {
        _consumer.Received += OnMessageReceived;
        
        // 开始监听队列
        _listenResult = await _consumer.ListenAsync("my_queue", options =>
        {
            options.AutoAck = false; // 手动确认
            options.Durable = true;
            options.FetchCount = 10; // 每次获取10条消息
        });
    }
    
    private void OnMessageReceived(RecieveResult result)
    {
        try
        {
            _logger.LogInformation($"Received message: {result.Body}");
            // 处理消息...
            
            // 确认消息处理完成
            result.Commit();
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Failed to process message");
            // 不调用Commit()，消息会被重新入队
        }
    }
    
    public Task StopAsync(CancellationToken cancellationToken)
    {
        _listenResult?.Stop();
        _consumer.Received -= OnMessageReceived;
        return Task.CompletedTask;
    }
    
    public void Dispose()
    {
        _listenResult?.Stop();
    }
}

其他用法

//批量获取消息
var message = await _consumer.ReceiveGetAsync("my_queue", new ConsumeQueueOptions
{
    FetchCount = 5
});

if (!string.IsNullOrEmpty(message))
{
    Console.WriteLine($"Received: {message}");
}


//获取队列消息数量
var count = await _consumer.GetMQMessageCountAsync("my_queue");
Console.WriteLine($"Messages in queue: {count}");


//使用交换机和路由键
// 生产者
await _producer.PublishAsync("my_exchange", "my.routing.key", "Hello RabbitMQ", options =>
{
    options.Type = RabbitMQExchangeType.Topic;
    options.Durable = true;
});

// 消费者
await _consumer.ListenAsync("my_exchange", "my_queue", options =>
{
    options.RoutingKeys = new[] { "my.*" }; // 订阅所有以my.开头的路由键
    options.AutoAck = false;
});