C#中RabbitMQ的使用小结

木太 · 发表于 2025-12-17 11:25:14

一、RabbitMQ是什么？

RabbitMQ是一个开源的消息代理软件，实现了AMQP（高级消息队列协议），用来实现应用程序之间的异步通信。简单说，它就像一个"消息邮局"，生产者把消息投递到"邮局"，消费者从"邮局"取走消息。

RabbitMQ：企业级消息中间件

RabbitMQ是一个开源的消息队列中间件，作用是：

✅ 系统解耦：不同应用间通过消息传递通信，无需直接依赖 ✅ 异步处理：将耗时操作放入队列，提升系统响应速度 ✅ 流量削峰：在高并发场景下缓冲请求，避免系统崩溃 ✅ 高可用：支持镜像队列和仲裁队列，保证服务可用性

典型应用：电商平台用户注册后，通过RabbitMQ异步发送注册邮件和短信；电商大促时，用消息队列缓冲订单，避免下游系统被压垮。

RabbitMQ是用Erlang语言写的，但C#有很好的客户端库支持，特别适合企业级应用。

二、环境准备

2.1. 安装RabbitMQ服务器

Windows：下载安装RabbitMQ（官网下载），安装时记得同时安装Erlang
Linux：使用yum或apt安装，或者用Docker

重要提示：安装完成后，启用Web管理插件（rabbitmq-plugins enable rabbitmq_management），然后访问http://localhost:15672/，默认账号guest/guest

2.2. 安装C#客户端库

# 使用NuGet安装RabbitMQ.Client（官方客户端）
Install-Package RabbitMQ.Client
 
# 如果喜欢更简单的封装，可以安装EasyNetQ（基于RabbitMQ.Client）
Install-Package EasyNetQ

三、基础使用示例

3.1. 简单发布/订阅模式（使用RabbitMQ.Client）

发布者（Publisher）：

using RabbitMQ.Client;
using System.Text;

class Program
{
    static void Main(string[] args)
    {
        var factory = new ConnectionFactory() { HostName = "localhost" };
        using (var connection = factory.CreateConnection())
        using (var channel = connection.CreateModel())
        {
            // 声明交换机（使用fanout类型，适合广播）
            channel.ExchangeDeclare(exchange: "logs", type: "fanout");
            
            // 发送消息
            string message = "Hello RabbitMQ from C#!";
            var body = Encoding.UTF8.GetBytes(message);
            
            channel.BasicPublish(exchange: "logs", 
                                routingKey: "", 
                                basicProperties: null, 
                                body: body);
            
            Console.WriteLine($" [x] Sent '{message}'");
        }
    }
}

订阅者（Consumer）：

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

class Program
{
    static void Main(string[] args)
    {
        var factory = new ConnectionFactory() { HostName = "localhost" };
        using (var connection = factory.CreateConnection())
        using (var channel = connection.CreateModel())
        {
            // 声明相同交换机
            channel.ExchangeDeclare(exchange: "logs", type: "fanout");
            
            // 创建临时队列（会自动删除）
            var queueName = channel.QueueDeclare().QueueName;
            
            // 绑定队列到交换机
            channel.QueueBind(queue: queueName,
                             exchange: "logs",
                             routingKey: "");
            
            // 创建消费者
            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: queueName,
                                autoAck: true,
                                consumer: consumer);
            
            Console.WriteLine("  Waiting for messages. To exit press CTRL+C");
            Console.ReadLine();
        }
    }
}

四、高级特性

4.1. 持久化消息（确保消息不丢失）

重要：必须同时设置交换机、队列、消息为持久化，才能保证消息不丢失

// 发布者
channel.ExchangeDeclare(exchange: "persistent", type: "direct", durable: true);
channel.QueueDeclare(queue: "persistent_queue", durable: true);
channel.QueueBind(queue: "persistent_queue", exchange: "persistent", routingKey: "key");

var message = "Persistent message";
var body = Encoding.UTF8.GetBytes(message);
var properties = channel.CreateBasicProperties();
properties.Persistent = true; // 关键！设置为持久化

channel.BasicPublish(exchange: "persistent", 
                    routingKey: "key",
                    basicProperties: properties,
                    body: body);

4.2. 使用EasyNetQ简化代码

EasyNetQ是RabbitMQ.Client的封装，让代码更简洁：

// 安装EasyNetQ
// Install-Package EasyNetQ

using EasyNetQ;
using System;

class Program
{
    static void Main()
    {
        // 创建连接
        var bus = RabbitHutch.CreateBus("host=localhost");
        
        // 发布消息
        bus.Publish(new Message { Text = "Hello EasyNetQ!" });
        
        // 订阅消息
        bus.Subscribe<Message>("my-queue", message => 
        {
            Console.WriteLine($"Received: {message.Text}");
        });
        
        Console.WriteLine("Waiting for messages...");
        Console.ReadLine();
    }
}

五、实际应用场景

5.1.电商订单处理（解耦系统）

// 订单服务（生产者）
public void CreateOrder(Order order)
{
    // 处理订单逻辑...
    
    // 发布订单消息到RabbitMQ
    var bus = RabbitHutch.CreateBus("host=localhost");
    bus.Publish(new OrderCreatedEvent { OrderId = order.Id });
}

// 邮件服务（消费者）
bus.Subscribe<OrderCreatedEvent>("order-events", order => 
{
    // 发送确认邮件
    EmailService.SendOrderConfirmation(order.OrderId);
});

5.2.日志收集系统

// 日志服务（生产者）
public void Log(string message)
{
    var bus = RabbitHutch.CreateBus("host=localhost");
    bus.Publish(new LogMessage { Message = message, Timestamp = DateTime.UtcNow });
}

// 日志处理服务（消费者）
bus.Subscribe<LogMessage>("log-queue", log => 
{
    // 保存到数据库或文件
    LogRepository.Save(log);
});

六、专业建议

连接管理：使用连接池，避免频繁创建连接

// 使用连接工厂创建连接
var factory = new ConnectionFactory { HostName = "localhost" };
using (var connection = factory.CreateConnection())
{
    // 用同一个连接创建多个channel
    using (var channel = connection.CreateModel())
    {
        // ...
    }
}

错误处理：不要只检查连接状态，要处理所有可能的异常

try
{
    // 消息处理逻辑
}
catch (Exception ex)
{
    // 记录错误并重试
    Console.WriteLine($"处理消息失败: {ex.Message}");
}

队列持久化：生产环境中，几乎所有的队列都应该是持久化的

channel.QueueDeclare(queue: "my-queue", 
                   durable: true, 
                   exclusive: false, 
                   autoDelete: false);

消息确认：使用手动确认（manual acknowledgment）确保消息被正确处理

channel.BasicConsume(queue: "my-queue",
                   autoAck: false, // 关键！设置为false
                   consumer: consumer);

七、常见问题

Q: RabbitMQ和MQTT有什么区别？ A: RabbitMQ是企业级消息中间件，适合应用间通信；MQTT是物联网专用轻量级协议，适合设备间通信。两者可以结合使用，例如用MQTT收集物联网设备数据，用RabbitMQ处理业务逻辑。

Q: 为什么我的消息不见了？ A: 可能原因：

队列没有持久化
没有正确绑定交换机和队列
消费者没有正确订阅
交换机类型不匹配（如用direct交换机但用fanout方式发送）

Q: 如何监控RabbitMQ？ A: 用Web管理界面（http://localhost:15672），或者用RabbitMQ的API进行监控。

八、其他类似中间件对比总结表

九、选择建议

如果你需要低延迟、简单路由 → 选RabbitMQ
如果你处理海量数据、日志收集 → 选Kafka
如果你需要顺序消息、电商场景 → 选RocketMQ
如果你需要多协议支持、复杂路由 → 选ActiveMQ
如果你追求极致速度、不关心数据丢失 → 选ZeroMQ
如果你做云原生应用、需要长期存储 → 选Pulsar

到此这篇关于C#中RabbitMQ的使用小结的文章就介绍到这了,更多相关C# RabbitMQ使用内容请搜索琼殿技术社区以前的文章或继续浏览下面的相关文章希望大家以后多多支持琼殿技术社区！

您可能感兴趣的文章:

在C# .NET中使用RabbitMQ实现发布/订阅模式的方法
C#使用RabbitMQ的详细教程
C#使用RabbitMQ发送和接收消息工具类的实现
C# RabbitMQ的使用详解
C#通过rabbitmq实现定时任务(延时队列)
C#用RabbitMQ实现消息订阅与发布
C#利用RabbitMQ实现点对点消息传输
c# rabbitmq 简单收发消息的示例代码
C#操作RabbitMQ的完整实例

[教程] C#中RabbitMQ的使用小结

目录