### 1、RabbitMQ.MessageProducer
> 新建类 `ConsumptionACKConfirm`



```csharp
public static void Show()
{
    var factory = new ConnectionFactory()
    {
        HostName = "localhost",
        UserName = "guest",
        Password = "guest"
    };

    using var connection = factory.CreateConnection();
    using var channel = connection.CreateModel();

    Console.WriteLine("生产者准备就绪");

    channel.QueueDeclare("ConsumptionACKConfirmQueue", true, false, false, null);
    channel.ExchangeDeclare("ConsumptionACKConfirmQueueExchange", ExchangeType.Direct, true, false);
    channel.QueueBind("ConsumptionACKConfirmQueue", "ConsumptionACKConfirmQueueExchange", "ConsumptionACKConfirmKey");

    // 设置优先级
    var props = channel.CreateBasicProperties();

    for (int i = 1; i < 1000; i++)
    {
        props.DeliveryMode = 2;

        string message = $"消息{i}";
        channel.BasicPublish("ConsumptionACKConfirmQueueExchange", "ConsumptionACKConfirmKey", props, Encoding.UTF8.GetBytes(message));

        Thread.Sleep(300);
        Console.WriteLine($"【{message}】 已发送");
    }

}
```
### 2、RabbitMQ.MessageConsumer01
> 新建类 `ConsumptionACKConfirm`

```csharp
public static void Show()
 {
     var factory = new ConnectionFactory()
     {
         HostName = "localhost",
         UserName = "guest",
         Password = "guest"
     };

     using var connection = factory.CreateConnection();
     using var channel = connection.CreateModel();

     #region 定义消费者

     var consumer = new EventingBasicConsumer(channel);

     int i = 0;

     consumer.Received += (model, ea) =>
     {
         var message = Encoding.UTF8.GetString(ea.Body.ToArray());
         //Console.WriteLine($"接受到消息：{message}");

         if (i < 50)
         {
             // 手动确认 消息正常消费，告诉Broker，可以将当前消息删除
             channel.BasicAck(ea.DeliveryTag, false);
             Console.WriteLine(message);
         }
         else
         {
             // 否定，告诉 Broker，消息没有正常消费，重新写入到队列中
             channel.BasicReject(ea.DeliveryTag, true);
         }
         i++;
     };

     Console.WriteLine("消费者准备就绪");
     {
         // 处理消息
         //autoAsk true 自动确认
         //channel.BasicConsume("ConsumptionACKConfirmQueue", true, consumer);
     }
     {
         // 处理消息
         //autoAsk false 手动（显示）确认
         channel.BasicConsume("ConsumptionACKConfirmQueue", false, consumer);
     }

     Console.ReadKey();

     #endregion
 }
```
### 3、注意
#### 1、自动消费
`BasicConsume` 的第二个参数设置为 true，会在初次进入接收消息时，`RabbitMQ` 网页客户端会清空消息，会将所有消息交由消费端去处理，无法保证消息处理的正确性
#### 2、手动消费
> 需要将 `BasicConsume` 的第二个参数设置为 false，使用 `channel.BasicAck(ea.DeliveryTag, false)` 进行手动确认，若确认失败，使用 `channel.BasicReject(ea.DeliveryTag, true)` 重新将消息写入队列

6、使用 RabbitMQ 实现手动确认消息消费

### 1、TX事务模式
> **RabbitMQ.MessageProducer**
> 新建类 `ProductionMessageTx`



```csharp
public static void Show()
{
    var factory = new ConnectionFactory()
    {
        HostName = "localhost",
        UserName = "guest",
        Password = "guest"
    };

    using var connection = factory.CreateConnection();
    using var channel = connection.CreateModel();

    Console.WriteLine("生产者准备就绪");

    channel.QueueDeclare("MessageTxQueue01", true, false, false, null);
    channel.QueueDeclare("MessageTxQueue02", true, false, false, null);

    channel.ExchangeDeclare("MessageTxQueueExchange", ExchangeType.Direct, true, false, null);

    channel.QueueBind("MessageTxQueue01", "MessageTxQueueExchange", "MessageTxKey01");
    channel.QueueBind("MessageTxQueue02", "MessageTxQueueExchange", "MessageTxKey02");

    string message = "";
	int i = 0;
    while (!message.Equals("quit", StringComparison.CurrentCultureIgnoreCase))
    {
        message = Console.ReadLine();
        var body = Encoding.UTF8.GetBytes(message);
        try
        {
            // 开启事务机制
            channel.TxSelect();

            // 发送消息
            channel.BasicPublish("MessageTxQueueExchange", "MessageTxKey01", null, body);
            
            if (i == 2)
                throw new Exception("发送消息异常");
            
            channel.BasicPublish("MessageTxQueueExchange", "MessageTxKey02", null, body);

            // 事务提交
            channel.TxCommit();
            i++;
            Console.WriteLine($"{message} 发送到Broke成功");
        }
        catch (Exception)
        {
            Console.WriteLine($"{message} 发送到Broke失败");
            channel.TxRollback();
        }
    }
    Console.Read();
}
```

### 2、`Confirm` 模式
> **RabbitMQ.MessageProducer**
> 新建类 `ProductionMessageConfirm`

```csharp
public static void Show()
{

    var factory = new ConnectionFactory()
    {
        HostName = "localhost",
        UserName = "guest",
        Password = "guest"
    };

    using var connection = factory.CreateConnection();
    using var channel = connection.CreateModel();

    Console.WriteLine("生产者准备就绪");

    channel.QueueDeclare("ConfirmSelectQueue", true, false, false, null);
    channel.ExchangeDeclare("ConfirmSelectQueueExchange", ExchangeType.Direct, true, false, null);

    channel.ExchangeBind("ConfirmSelectQueue", "ConfirmSelectQueueExchange", "ConfirmSelectKey");
    string message = "";
    while (!message.Equals("quit", StringComparison.CurrentCultureIgnoreCase))
    {
        message = Console.ReadLine();
        var body = Encoding.UTF8.GetBytes(message);
        try
        {
            // 开启事务机制
            channel.ConfirmSelect();

            // 发送消息
            channel.BasicPublish("ConfirmSelectQueueExchange", "ConfirmSelectKey", null, body);

            // 事务提交
            if (channel.WaitForConfirms())
                Console.WriteLine($"{message} 发送到Broke成功");

            // 所有消息发送成功，正常执行，有消息发送失败，抛出异常
            // channel.WaitForConfirmsOrDie();

        }
        catch (Exception)
        {
            Console.WriteLine($"{message} 发送到Broke失败");
            // 通知管理员，写入日志，重试
        }
    }
    Console.Read();
}
```

5、使用 RabbitMQ 实现消息的事务机制

### 1、RabbitMQ.MessageProducer
> 新建类 `**PriorityQueue**`



```csharp
public static void Show()
{
    var factory = new ConnectionFactory()
    {
        HostName = "localhost",
        UserName = "guest",
        Password = "guest"
    };

    using var connection = factory.CreateConnection();
    using var channel = connection.CreateModel();

    channel.QueueDeclare("PriorityQueue", true, false, false, new Dictionary<string, object>()
    {
        // 指定队列要支持优先级队列
        { "x-max-priority",10 }
    });
    // 声明交换机
    channel.ExchangeDeclare("PriorityQueueExchange", ExchangeType.Direct, true, false, null);
    // 绑定交换机和队列
    channel.QueueBind("PriorityQueue", "PriorityQueueExchange", "PriorityKey");
    // 一些待发送的消息
    string[] questionList = { "vip1 来请教", "甲 同学来请教", "乙 同学来请教", "丙 同学来请教", "丁 同学来请教", "vip2 来请教" };

    var props = channel.CreateBasicProperties();
    // 设置持久化
    props.DeliveryMode = 2;
    foreach (var msg in questionList)
    {
        if (msg.StartsWith("vip"))
        {
            // 设置优先级
            props.Priority = 9;
            channel.BasicPublish("PriorityQueueExchange", "PriorityKey", props, Encoding.UTF8.GetBytes(msg));
        }
        else
        {
            props.Priority = 1;
            channel.BasicPublish("PriorityQueueExchange", "PriorityKey", props, Encoding.UTF8.GetBytes(msg));
        }
        Console.WriteLine($"{msg} 已发送");
    }

    Console.ReadLine();

}
```


### 2、RabbitMQ.MessageConsumer01
> 新建类 `**PriorityQueue**`

```csharp
public static void Show()
{
    var factory = new ConnectionFactory()
    {
        HostName = "localhost",
        UserName = "guest",
        Password = "guest"
    };

    using var connection = factory.CreateConnection();
    using var channel = connection.CreateModel();

    var consumer = new EventingBasicConsumer(channel);
    consumer.Received += (model, ea) =>
    {
        Console.WriteLine(Encoding.UTF8.GetString(ea.Body.ToArray()));
    };
    Console.WriteLine("消费者准备就绪。。。");
    // 处理消息
    channel.BasicConsume("PriorityQueue", true, consumer);

    Console.ReadLine();
}
```

![image.png](https://cdn.netclub.top/images/0287b6edfb2f810eb93bd6a61e481d82.image.webp?imageslim)

### 3、注意

> 优先级

```csharp
// 声明的队列需要加入参数 x-max-priority
var props = channel.CreateBasicProperties();
// 值越大，优先级越高
props.Priority = 9;
```

> 持久化

![image.png](https://cdn.netclub.top/images/91f41074a0c72fa290acb68b87de5630.image.webp?imageslim)

```csharp
// 在声明队列和交换机时需要将 durable 设置为 true
var props = channel.CreateBasicProperties();
// 设置持久化
props.DeliveryMode = 2;
```

4、使用 RabbitMQ 设置消息优先级、持久化

### 1、新建项目
> 在上篇的基础上新建控制台

```csharp
donnet new console -n RabbitMQ.MessageConsumer02 -f net5.0

dotnet sln add .\RabbitMQ.MessageConsumer02
```



### 2、RabbitMQ.MessageProducer
> 新建类 `FanoutExchange`

```csharp
public static void Show()
{

    var factory = new ConnectionFactory()
    {
        HostName = "localhost",
        UserName = "guest",
        Password = "guest"
    };

    using var connection = factory.CreateConnection();
    using var channel = connection.CreateModel();

    // 定义两个队列
    channel.QueueDeclare("FanoutExchange01", true, false, false, null);
    channel.QueueDeclare("FanoutExchange02", true, false, false, null);
	
    // 定义交换机名称，交换机类型
    channel.ExchangeDeclare("FanoutExchange", ExchangeType.Fanout, true, false, null);
	
    // 将两个队列绑定到交换机上
    channel.QueueBind("FanoutExchange01", "FanoutExchange", string.Empty, null);
    channel.QueueBind("FanoutExchange02", "FanoutExchange", string.Empty, null);

    // 在控制台输入消息，按enter键发送消息
    int i = 1;
    while (true)
    {
        var message = $"通知{i}";
        var body = Encoding.UTF8.GetBytes(message);
        // 基本发布
        channel.BasicPublish("FanoutExchange", string.Empty, null, body);

        Console.WriteLine($"通知【{message}】已发送到队列");

        Thread.Sleep(2000);
        i++;

    }

}
```
### 3、RabbitMQ.MessageConsumer01
> 新建类 `FanoutExchange`

```csharp
public static void Show()
 {

     var factory = new ConnectionFactory()
     {
         HostName = "localhost",
         UserName = "guest",
         Password = "guest"
     };

     using var connection = factory.CreateConnection();
     using var channel = connection.CreateModel();

     // 声明交换机
     channel.ExchangeDeclare("FanoutExchange", ExchangeType.Fanout, true, false, null);

     // 声明队列
     channel.QueueDeclare("FanoutExchange01", true, false, false, null);

     // 绑定 exchange 和 queue
     channel.QueueBind("FanoutExchange01", "FanoutExchange", string.Empty, null);

     // 定义消费者
     var consumer = new EventingBasicConsumer(channel);
     consumer.Received += (model, ea) =>
     {
         var body = ea.Body;
         var message = Encoding.UTF8.GetString(body.ToArray());

         Console.WriteLine($"接收成功!【{message}】，邮件通知");

     };
     Console.WriteLine("通知服务准备就绪");

     string result = channel.BasicConsume("FanoutExchange01", true, consumer);
     Console.WriteLine(result);
     
     Console.ReadLine();

 }
```
### 4、RabbitMQ.MessageConsumer02
> 新建类 `FanoutExchange`

```csharp
public static void Show()
{

    var factory = new ConnectionFactory()
    {
        HostName = "localhost",
        UserName = "guest",
        Password = "guest"
    };

    using var connection = factory.CreateConnection();
    using var channel = connection.CreateModel();

    // 声明交换机
    channel.ExchangeDeclare("FanoutExchange", ExchangeType.Fanout, true, false, null);

    // 声明队列
    channel.QueueDeclare("FanoutExchange02", true, false, false, null);

    // 绑定 exchange 和 queue
    channel.QueueBind("FanoutExchange02", "FanoutExchange", string.Empty, null);

    // 定义消费者
    var consumer = new EventingBasicConsumer(channel);
    consumer.Received += (model, ea) =>
    {
        var body = ea.Body;
        var message = Encoding.UTF8.GetString(body.ToArray());

        Console.WriteLine($"接收成功!【{message}】，邮件通知");

    };
    Console.WriteLine("通知服务准备就绪");

    channel.BasicConsume("FanoutExchange02", true, consumer);

    Console.ReadLine();

}
```
### 5、运行
> 两个队列都能接收到消息并消费


![image.png](https://cdn.netclub.top/images/2d018af26ee1f7adbe611c1b361c828a.image.webp?imageslim)

3、使用 RabbitMQ 实现发布订阅功能

> 环境：
> **Visual Studio 2022**
> **NET5**
> **RabbitMQ.Client 6.2.2**

> 1、新建两个控制台系统
```csharp
dotnet new sln
dotnet new console -n RabbitMQ.MessageProducer -f net5.0
donnet new console -n RabbitMQ.MessageConsumer01 -f net5.0

dotnet sln add .\RabbitMQ.MessageProducer
dotnet sln add .\RabbitMQ.MessageProducer
```



> 安装 `**RabbitMQ**`

```csharp
<PackageReference Include="RabbitMQ.Client" Version="6.2.2" />
```
> `**RabbitMQ.MessageProducer**`
> `MessageProducer` 生产消息

```csharp
# 新建类 ProductionConsumer
public static void Show()
{
    var factory = new ConnectionFactory()
    {
        HostName = "localhost",
        UserName = "guest",
        Password = "guest"
    };

    using var connection = factory.CreateConnection();
    using var channel = connection.CreateModel();
	
    // 队列名称OnlyProducerMessage
    channel.QueueDeclare(queue: "OnlyProducerMessage", durable: true, exclusive: false, autoDelete: false, arguments: null);
	// 交换机名称 Exchange
    channel.ExchangeDeclare(exchange: "OnlyProducerMessageExChange", type: ExchangeType.Direct, durable: true, autoDelete: false, arguments: null);
	// 将队列绑定到交换机
    channel.QueueBind(queue: "OnlyProducerMessage", exchange: "OnlyProducerMessageExChange", routingKey: string.Empty, arguments: null);

    Console.ForegroundColor = ConsoleColor.Red;

    Console.WriteLine("生产者ProducerDemo已准备就绪");

    int i = 1;
    {
        while (true)
        {
            string message = $"消息{i}";
            byte[] body = Encoding.UTF8.GetBytes(message);
            channel.BasicPublish(exchange: "OnlyProducerMessageExChange", routingKey: string.Empty, basicProperties: null, body: body);
            Console.WriteLine($"消息：{message}已发送");
            i++;
            Thread.Sleep(200);
        }
    }
}
```
> `**RabbitMQ.MessageConsumer01**`
> `ProductionConsumer` 消费消息

```csharp
public static void Show()
{
    var factory = new ConnectionFactory()
    {
        HostName = "localhost",
        UserName = "guest",
        Password = "guest"
    };

    using var connection = factory.CreateConnection();
    using var channel = connection.CreateModel();

    Console.ForegroundColor = ConsoleColor.Green;
    try
    {
        var consumer = new EventingBasicConsumer(channel);
        consumer.Received += (model, ea) =>
        {
            var message = Encoding.UTF8.GetString(ea.Body.ToArray());
            Console.WriteLine($"消费者01 接受消息：{message}");
        };
        channel.BasicConsume(queue: "OnlyProducerMessage", autoAck: true, consumer: consumer);
        Console.WriteLine("Press [enter] to exit");
        Console.ReadLine();
    }
    catch (Exception ex)
    {
        Console.WriteLine(ex.Message);
    }

}
```
> 分别在 `Program` 中调用编译运行
> 访问`http://127.0.0.1:15672`
> 初始账号密码
> `guest` `guest`

![image.png](https://cdn.netclub.top/images/e027186ae10fd9c43f30d50f9ca59890.image.webp?imageslim)