C#中RabbitMQ的使用小结

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

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}'");
        }
    }
}

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();
        }
    }
}

// 发布者
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);

// 安装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();
    }
}

// 订单服务（生产者）
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);
});

// 日志服务（生产者）
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);

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