在C# .net中RabbitMQ的核心类型和属性，除了交换机，队列关键的类型 / 属性，影响其行为

前言：本文详细介绍了在C#.NET中使用RabbitMQ的核心组件与实现方法。主要内容包括：1) RabbitMQ四种交换机类型(Direct/Fanout/Topic/Headers)的实现原理及C#代码示例；2) 队列的关键属性配置(持久化、排他队列、自动删除等)及其应用场景；3) 高级队列特性如死信队列、消息TTL、优先级队列的实现方式。文章通过完整可运行的代码示例，展示了如何在.NET环境中利用RabbitMQ.Client库构建可靠的消息系统，并针对不同业务场景提供了最佳实践建议，如订单处理、广播通知等典型应用。

        在C# .net中RabbitMQ 中不同的交换机（Exchange）类型，这是决定消息如何路由到队列的关键组件，除了交换机，队列也有一些关键的类型 / 属性，影响其行为，我会一并说明。

        在 C# .NET 开发环境中，如何使用 RabbitMQ 的各种核心类型（主要是交换机类型），我会结合 .NET 生态下最常用的 RabbitMQ.Client 库，为你提供可直接运行的代码示例和清晰的解释。

        在 .NET 中使用 RabbitMQ，核心依然是操作 四种交换机类型（Direct、Fanout、Topic、Headers），API 调用方式与其他语言逻辑一致，但语法符合 C# 规范。以下是完整的实现示例，基于 RabbitMQ.Client NuGet 包（这是官方推荐的 .NET 客户端）。

前置准备

首先需要安装依赖包：

# 通过 NuGet 安装 RabbitMQ.Client
Install-Package RabbitMQ.Client
# 或使用 .NET CLI
dotnet add package RabbitMQ.Client

一、通用连接工具类

先封装一个通用的连接创建方法，避免重复代码：

using RabbitMQ.Client;
using System;

public static class RabbitMqHelper
{
// 创建 RabbitMQ 连接
public static IConnection CreateConnection()
{
var factory = new ConnectionFactory
{
HostName = "localhost", // RabbitMQ 服务器地址
UserName = "guest", // 默认用户名
Password = "guest", // 默认密码
Port = 5672 // 默认端口
};

try
{
var connection = factory.CreateConnection();
Console.WriteLine("RabbitMQ 连接成功！");
return connection;
}
catch (Exception ex)
{
Console.WriteLine($"连接失败：{ex.Message}");
throw;
}
}
}

二、核心：各交换机（Exchange）类型实现示例

交换机是接收生产者消息并将其路由到队列的组件，它的类型决定了路由规则。

1. Direct Exchange（直连交换机）

• 核心逻辑：RoutingKey 与 BindingKey 完全匹配时路由消息。
• 适用场景：一对一的消息发送（比如任务分发、精准通知）。
• 示例代码： public class DirectExchangeDemo
  {
  // 生产者：发送消息到直连交换机
  public static void SendDirectMessage(string routingKey, string message)
  {
  using (var connection = RabbitMqHelper.CreateConnection())
  using (var channel = connection.CreateModel())
  {
  // 1. 声明直连交换机（durable: true 表示持久化）
  channel.ExchangeDeclare(
  exchange: "direct_exchange_demo",
  type: ExchangeType.Direct, // 直连类型
  durable: true);

  // 2. 声明队列（持久化）
  channel.QueueDeclare(
  queue: "direct_queue_demo",
  durable: true,
  exclusive: false,
  autoDelete: false,
  arguments: null);

  // 3. 绑定交换机和队列（指定 BindingKey）
  channel.QueueBind(
  queue: "direct_queue_demo",
  exchange: "direct_exchange_demo",
  routingKey: routingKey);

  // 4. 发送消息（指定 RoutingKey）
  var body = System.Text.Encoding.UTF8.GetBytes(message);
  channel.BasicPublish(
  exchange: "direct_exchange_demo",
  routingKey: routingKey,
  basicProperties: null,
  body: body);

  Console.WriteLine($"已发送 Direct 消息：{message} (RoutingKey: {routingKey})");
  }
  }

  // 消费者：接收直连交换机的消息
  public static void ReceiveDirectMessage(string routingKey)
  {
  using (var connection = RabbitMqHelper.CreateConnection())
  using (var channel = connection.CreateModel())
  {
  // 重复声明交换机/队列（幂等操作，不存在则创建）
  channel.ExchangeDeclare("direct_exchange_demo", ExchangeType.Direct, true);
  channel.QueueDeclare("direct_queue_demo", true, false, false, null);
  channel.QueueBind("direct_queue_demo", "direct_exchange_demo", routingKey);

  // 创建消费者
  var consumer = new EventingBasicConsumer(channel);
  consumer.Received += (model, ea) =>
  {
  var body = ea.Body.ToArray();
  var message = System.Text.Encoding.UTF8.GetString(body);
  Console.WriteLine($"已接收 Direct 消息：{message} (RoutingKey: {ea.RoutingKey})");

  // 手动确认消息已处理
  channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
  };

  // 开始消费
  channel.BasicConsume(
  queue: "direct_queue_demo",
  autoAck: false, // 关闭自动确认，手动 Ack
  consumer: consumer);

  Console.WriteLine("Direct 消费者已启动，按任意键退出…");
  Console.ReadKey();
  }
  }
  }

2. Fanout Exchange（扇出 / 广播交换机）

• 核心逻辑：无视 routing key 和 binding key，将消息广播到所有绑定到该交换机的队列。
• 适用场景：一对多的消息广播（比如实时通知、状态同步）。
• 特点：性能最高，因为无需匹配逻辑。
• 示例代码： public class FanoutExchangeDemo
  {
  public static void SendFanoutMessage(string message)
  {
  using (var connection = RabbitMqHelper.CreateConnection())
  using (var channel = connection.CreateModel())
  {
  // 声明扇出交换机
  channel.ExchangeDeclare(
  exchange: "fanout_exchange_demo",
  type: ExchangeType.Fanout, // 扇出类型
  durable: true);

  // 声明两个队列并绑定到交换机（无需指定 RoutingKey）
  channel.QueueDeclare("fanout_queue_1", true, false, false, null);
  channel.QueueDeclare("fanout_queue_2", true, false, false, null);
  channel.QueueBind("fanout_queue_1", "fanout_exchange_demo", "");
  channel.QueueBind("fanout_queue_2", "fanout_exchange_demo", "");

  // 发送消息（RoutingKey 为空）
  var body = System.Text.Encoding.UTF8.GetBytes(message);
  channel.BasicPublish("fanout_exchange_demo", "", null, body);
  Console.WriteLine($"已发送 Fanout 广播消息：{message}");
  }
  }

  // 消费 fanout_queue_1 的消息
  public static void ReceiveFanoutMessage1()
  {
  using (var connection = RabbitMqHelper.CreateConnection())
  using (var channel = connection.CreateModel())
  {
  channel.ExchangeDeclare("fanout_exchange_demo", ExchangeType.Fanout, true);
  channel.QueueDeclare("fanout_queue_1", true, false, false, null);
  channel.QueueBind("fanout_queue_1", "fanout_exchange_demo", "");

  var consumer = new EventingBasicConsumer(channel);
  consumer.Received += (model, ea) =>
  {
  var message = System.Text.Encoding.UTF8.GetString(ea.Body.ToArray());
  Console.WriteLine($"队列1 接收 Fanout 消息：{message}");
  channel.BasicAck(ea.DeliveryTag, false);
  };

  channel.BasicConsume("fanout_queue_1", false, consumer);
  Console.WriteLine("Fanout 消费者1已启动…");
  Console.ReadKey();
  }
  }
  }

3. Topic Exchange（主题交换机）

• 核心逻辑：基于通配符匹配 routing key 和 binding key，支持模糊路由。
  • *：匹配一个单词（比如 order.* 匹配 order.create，但不匹配 order.create.user）。
  • #：匹配零个或多个单词（比如 order.# 匹配 order.create、order.create.user）。
• 适用场景：多维度的消息路由（比如按业务模块 + 操作类型路由）。
• 示例代码： public class TopicExchangeDemo
  {
  public static void SendTopicMessage(string routingKey, string message)
  {
  using (var connection = RabbitMqHelper.CreateConnection())
  using (var channel = connection.CreateModel())
  {
  // 声明主题交换机
  channel.ExchangeDeclare(
  exchange: "topic_exchange_demo",
  type: ExchangeType.Topic, // 主题类型
  durable: true);

  // 绑定规则：
  // queue_order 绑定 "order.*" → 匹配 order.create、order.pay 等
  // queue_error 绑定 "#.error" → 匹配 order.error、user.login.error 等
  channel.QueueDeclare("queue_order", true, false, false, null);
  channel.QueueDeclare("queue_error", true, false, false, null);
  channel.QueueBind("queue_order", "topic_exchange_demo", "order.*");
  channel.QueueBind("queue_error", "topic_exchange_demo", "#.error");

  // 发送消息（比如 routingKey = "order.create" 或 "user.login.error"）
  var body = System.Text.Encoding.UTF8.GetBytes(message);
  channel.BasicPublish("topic_exchange_demo", routingKey, null, body);
  Console.WriteLine($"已发送 Topic 消息：{message} (RoutingKey: {routingKey})");
  }
  }
  }

4. Headers Exchange（头交换机）

• 核心逻辑：无视 routing key，通过消息的 header 属性（键值对）匹配路由规则。
• 匹配规则：通过消息头（Headers）而非 RoutingKey 匹配，支持 all/any 匹配规则。
• 适用场景：复杂的多条件路由（较少用，因为不如 topic 灵活）。
• 示例代码： public class HeadersExchangeDemo
  {
  public static void SendHeadersMessage(string message)
  {
  using (var connection = RabbitMqHelper.CreateConnection())
  using (var channel = connection.CreateModel())
  {
  // 声明头交换机
  channel.ExchangeDeclare(
  exchange: "headers_exchange_demo",
  type: ExchangeType.Headers, // 头类型
  durable: true);

  // 声明队列并绑定（匹配规则：type=payment 且 status=success）
  channel.QueueDeclare("headers_queue_demo", true, false, false, null);
  channel.QueueBind(
  queue: "headers_queue_demo",
  exchange: "headers_exchange_demo",
  routingKey: "", // 头交换机忽略 RoutingKey
  arguments: new Dictionary<string, object>
  {
  { "x-match", "all" }, // all=所有头匹配，any=任意一个匹配
  { "type", "payment" },
  { "status", "success" }
  });

  // 构建消息头并发送
  var properties = channel.CreateBasicProperties();
  properties.Headers = new Dictionary<string, object>
  {
  { "type", "payment" },
  { "status", "success" }
  };

  var body = System.Text.Encoding.UTF8.GetBytes(message);
  channel.BasicPublish("headers_exchange_demo", "", properties, body);
  Console.WriteLine($"已发送 Headers 消息：{message}");
  }
  }
  }

三、调用示例

// 测试 Direct 交换机
DirectExchangeDemo.SendDirectMessage("error", "系统错误日志");
DirectExchangeDemo.ReceiveDirectMessage("error");

// 测试 Fanout 交换机
FanoutExchangeDemo.SendFanoutMessage("全站广播通知");
FanoutExchangeDemo.ReceiveFanoutMessage1();

// 测试 Topic 交换机
TopicExchangeDemo.SendTopicMessage("order.create", "新订单创建");

// 测试 Headers 交换机
HeadersExchangeDemo.SendHeadersMessage("支付成功通知");

总结

四、队列的常见类型属性

除了交换机，队列也有一些关键的类型 / 属性，影响其行为：

队列的属性是在调用 QueueDeclare 方法时通过参数或 arguments 字典配置的，这些属性直接决定了队列的生命周期、消息可靠性、异常处理等核心行为。

前置说明

在 .NET 中，声明队列的核心方法是 IModel.QueueDeclare，基础签名如下：

QueueDeclareOk QueueDeclare(
string queue, // 队列名
bool durable, // 是否持久化
bool exclusive, // 是否排他
bool autoDelete, // 是否自动删除
IDictionary<string, object> arguments); // 扩展属性（死信、延迟、最大长度等）

下面按「基础属性」和「高级扩展属性」分类讲解，每个属性都包含作用说明 + C# 代码示例 + 适用场景。

一、队列基础属性（必配）

这是声明队列时的核心参数，决定队列的基础行为。

1. Durable（持久化）

• 作用：设置为 true 时，队列会被持久化到磁盘，RabbitMQ 重启后队列依然存在；false 则仅存于内存，重启后丢失。
• 注意：队列持久化 ≠ 消息持久化，需同时设置消息的 BasicProperties.Persistent = true 才能保证消息不丢。
• 代码示例：

using (var channel = connection.CreateModel())
{
// 声明持久化队列（核心：durable = true）
channel.QueueDeclare(
queue: "durable_queue",
durable: true, // 持久化
exclusive: false,
autoDelete: false,
arguments: null);

// 发送持久化消息（配合队列持久化）
var properties = channel.CreateBasicProperties();
properties.Persistent = true; // 消息持久化
channel.BasicPublish("", "durable_queue", properties, Encoding.UTF8.GetBytes("持久化消息"));
}

• 适用场景：生产环境核心业务队列（如订单、支付），必须保证重启后不丢失。

2. Exclusive（排他队列）

• 作用：设置为 true 时，队列仅对创建它的连接可见：
  • 只有创建队列的连接能访问该队列；
  • 连接关闭（或断开）后，队列会被自动删除；
  • 同一连接的多个信道（Channel）可共享该队列。
• 代码示例：

channel.QueueDeclare(
queue: "exclusive_queue",
durable: false, // 排他队列通常不持久化（因为生命周期和连接绑定）
exclusive: true, // 排他
autoDelete: false,
arguments: null);

• 适用场景：临时任务队列（如客户端专属的临时查询结果队列）、避免多客户端竞争的临时队列。

3. AutoDelete（自动删除）

• 作用：设置为 true 时，当最后一个消费者断开连接后，队列会被自动删除（无消费者时触发）。
• 注意：仅「消费者断开」触发，无消费者时创建队列不会立即删除。
• 代码示例：

channel.QueueDeclare(
queue: "auto_delete_queue",
durable: false,
exclusive: false,
autoDelete: true, // 自动删除
arguments: null);

• 适用场景：临时消费场景（如实时通知、一次性任务消费），避免无用队列堆积。

二、队列高级扩展属性（通过 arguments 配置）

这些属性通过 QueueDeclare 的 arguments 字典配置，是实现特殊功能的核心。

1. 死信队列（DLX，Dead-Letter-Exchange）

• 作用：当消息满足以下条件时，会被路由到「死信交换机」，进而转发到死信队列：
  • 消息被消费者拒绝（BasicNack/BasicReject）且 requeue = false；
  • 消息过期；
  • 队列达到最大长度，新消息挤掉旧消息。
• 核心参数：
  • x-dead-letter-exchange：死信交换机名称；
  • x-dead-letter-routing-key：死信消息的路由键（可选，默认用原消息的 routing key）。
• 代码示例：

// 1. 声明死信交换机和死信队列（普通的直连交换机即可）
channel.ExchangeDeclare("dlx_exchange", ExchangeType.Direct, true);
channel.QueueDeclare("dlx_queue", true, false, false, null);
channel.QueueBind("dlx_queue", "dlx_exchange", "dlx_routing_key");

// 2. 声明业务队列，并绑定死信属性
var arguments = new Dictionary<string, object>
{
{ "x-dead-letter-exchange", "dlx_exchange" }, // 绑定死信交换机
{ "x-dead-letter-routing-key", "dlx_routing_key" }, // 死信路由键
{ "x-message-ttl", 60000 } // 消息过期时间（60秒）
};

channel.QueueDeclare(
queue: "business_queue",
durable: true,
exclusive: false,
autoDelete: false,
arguments: arguments);

// 3. 模拟消息被拒绝，进入死信队列
channel.BasicReject(deliveryTag: 1, requeue: false); // requeue=false 才会进入死信

• 适用场景：处理失败消息（如订单支付超时、接口调用失败），避免消息丢失或无限重试。

2. 消息过期时间（x-message-ttl）

• 作用：设置队列中所有消息的默认过期时间（毫秒），超过时间未消费的消息会被丢弃或进入死信队列。
• 注意：也可给单个消息设置过期时间（BasicProperties.Expiration），优先级高于队列级别。
• 代码示例：

var arguments = new Dictionary<string, object>
{
{ "x-message-ttl", 30000 } // 队列中所有消息默认30秒过期
};
channel.QueueDeclare("ttl_queue", true, false, false, arguments);

// 单个消息设置过期时间（优先级更高）
var props = channel.CreateBasicProperties();
props.Expiration = "10000"; // 该消息10秒过期
channel.BasicPublish("", "ttl_queue", props, Encoding.UTF8.GetBytes("限时消息"));

• 适用场景：限时有效消息（如验证码、临时通知）。

3. 队列最大长度（x-max-length）

• 作用：限制队列最多存放的消息数量，超过后会删除旧消息（或发送到死信队列）。
• 可选参数：x-max-length-bytes（按字节数限制队列大小）。
• 代码示例：

var arguments = new Dictionary<string, object>
{
{ "x-max-length", 1000 }, // 队列最多存1000条消息
{ "x-dead-letter-exchange", "dlx_exchange" } // 超过数量的旧消息进入死信
};
channel.QueueDeclare("limit_queue", true, false, false, arguments);

• 适用场景：防止队列无限堆积导致内存 / 磁盘溢出（如日志队列）。

4. 队列过期时间（x-expires）

• 作用：设置队列的空闲过期时间（毫秒），当队列无消费者、无消息、无连接的时间超过该值，会被自动删除。
• 代码示例：

var arguments = new Dictionary<string, object>
{
{ "x-expires", 300000 } // 队列空闲5分钟后自动删除
};
channel.QueueDeclare("expire_queue", false, false, false, arguments);

• 适用场景：临时队列（如客户端临时订阅的队列），避免无用队列残留。

5. 最大优先级（x-max-priority）

• 作用：设置队列支持的消息优先级级别（0~ 指定值），优先级高的消息会先被消费。
• 代码示例：

var arguments = new Dictionary<string, object>
{
{ "x-max-priority", 10 } // 支持0-10级优先级（10最高）
};
channel.QueueDeclare("priority_queue", true, false, false, arguments);

// 发送带优先级的消息
var props = channel.CreateBasicProperties();
props.Priority = 9; // 该消息优先级为9（高优先级）
channel.BasicPublish("", "priority_queue", props, Encoding.UTF8.GetBytes("紧急消息"));

• 适用场景：区分消息紧急程度（如订单队列中，VIP 订单优先消费）。

三、完整示例：配置多属性的业务队列

运行

using RabbitMQ.Client;
using System;
using System.Collections.Generic;
using System.Text;

public class AdvancedQueueDemo
{
public static void CreateAdvancedQueue()
{
var factory = new ConnectionFactory { HostName = "localhost" };
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
// 1. 声明死信交换机和死信队列
channel.ExchangeDeclare("order_dlx_exchange", ExchangeType.Direct, true);
channel.QueueDeclare("order_dlx_queue", true, false, false, null);
channel.QueueBind("order_dlx_queue", "order_dlx_exchange", "order_dlx");

// 2. 声明订单队列，配置多属性
var queueArgs = new Dictionary<string, object>
{
{ "x-durable", true }, // 持久化
{ "x-message-ttl", 180000 }, // 消息3分钟过期
{ "x-max-length", 5000 }, // 队列最多存5000条消息
{ "x-dead-letter-exchange", "order_dlx_exchange" }, // 死信交换机
{ "x-dead-letter-routing-key", "order_dlx" }, // 死信路由键
{ "x-max-priority", 5 } // 5级优先级
};

channel.QueueDeclare(
queue: "order_business_queue",
durable: true,
exclusive: false,
autoDelete: false,
arguments: queueArgs);

Console.WriteLine("高级队列创建成功！");
}
}
}

总结

其他

RabbitMQ 安装卸载

https://blog.csdn.net/cao919/article/details/115602240

【电商 】订单减少库存业务流程，分布式ID策略选型，C# .net雪花ID代码，下单功能实现 ，异步延时队列

https://blog.csdn.net/cao919/article/details/126413455

SAAS多租户套餐权限模块功能按钮 设置 关键代码实现 JAVA C#

https://blog.csdn.net/cao919/article/details/143254585