C# 消息隊列demo

1.安裝完成rabbitMQ

參考上一篇如何安裝 rabbitMQ

2.安裝 OTP服務

https://www.erlang.org/downloads 

C#代碼

需要導入RabbitMQ.Client

在NuGet中搜索自取

隊列發送端

var factory = new ConnectionFactory();
//RabbitMQ伺服器地址
factory.HostName = "192.168.0.22";
//埠號
//factory.Port = 15672;
factory.UserName = "admin";
factory.Password = "123456";
//消息非同步轉發
factory.DispatchConsumersAsync = true;
//交換機名稱
string exchangeName = "exchange2";
//routingKey的值
string eventName = "key1";
//連接伺服器
using var conn = factory.CreateConnection();
int MsgInt = 1;

while (true)
{
    //待發送的消息
    string msg = $"{MsgInt++}條消息{DateTime.Now.ToLongTimeString()}";
    //創建通道
    using (var channel = conn.CreateModel())
    {
        //消息屬性
        var properties = channel.CreateBasicProperties();
        //傳輸模式，1：非持久化，2：持久化
        properties.DeliveryMode = 2;
        //聲明交換機
        channel.ExchangeDeclare(exchange: exchangeName, type: "direct");
        byte[] body = Encoding.UTF8.GetBytes(msg);
        //生產消息
        channel.BasicPublish(exchange: exchangeName, routingKey: eventName, mandatory: true, basicProperties: properties, body: body);
    }
    Console.WriteLine("發送消息：" + msg);
    Thread.Sleep(1000);
}

收信方

var factory = new ConnectionFactory();
//RabbitMQ伺服器地址
factory.HostName = "localhost";
//埠號
//factory.Port = 15672;
factory.UserName = "admin";
factory.Password = "123456";

//消息非同步轉發
factory.DispatchConsumersAsync = true;
//交換機名稱
string exchangeName = "exchange2";
//routingKey的值
string eventName = "key1";
using var conn = factory.CreateConnection();
//創建通道
using var channel = conn.CreateModel();

//聲明交換機
channel.ExchangeDeclare(exchange: exchangeName, type: "direct");
//聲明隊列
string queueName = "queue1";
channel.QueueDeclare(queueName, durable: true, exclusive: false, autoDelete: false, arguments: null);
//將隊列綁定到交換機中
channel.QueueBind(queueName, exchangeName, eventName);
//消費者拉取消息
AsyncEventingBasicConsumer consumer = new AsyncEventingBasicConsumer(channel);
consumer.Received += Consumer_Received;
//消費消息
channel.BasicConsume(queueName, autoAck: false, consumer);
Console.WriteLine("按回車退出");
Console.ReadLine();

async Task Consumer_Received(object sender, BasicDeliverEventArgs args)
{
    try
    {
        byte[] bytes = args.Body.ToArray();
        string msg = Encoding.UTF8.GetString(bytes);
        Console.WriteLine(DateTime.Now.ToLongTimeString() + "收到了信息" + msg);
        //消息確認處理，DeliveryTag：消息的編號 執行後等於告訴隊列 已經收到了消息
        channel.BasicAck(args.DeliveryTag, multiple: false);
        await Task.Delay(800);
    }
    catch (Exception ex)
    {
        //對沒有確認處理的消息進行消息重發
        channel.BasicReject(args.DeliveryTag, true);
        Console.WriteLine(DateTime.Now.ToLongTimeString() + "處理收到的消息出錯：" + ex);
    }
}

收信方和發送方 分別是 2個項目 做 測試

ConsoleApp1是發送 2是收信