.NET6中 RabbitMQ的使用

  
    Exe
    net6.0
    enable
    enable
  

  
    
	  
	  
	  
		  all
		  runtime; build; native; contentfiles; analyzers; buildtransitive
	  
	  
	  
	  
		  all
		  runtime; build; native; contentfiles; analyzers; buildtransitive
	  
	  
	  
	  
	  
	  
	  
	  
  

需要配置环境变量

erlanghome/bin

在安装目录

C:Program FilesRabbitMQ Serverrabbitmq_server-3.9.15sbin

下运行

rabbitmq-plugins enable rabbitmq_management

输入

rabbitmqctl status检查

服务-停止 rabbitmq

再在 目录下运行

rabbitmq-server.bat

再在server打开rabbitmq

浏览器输入

http://localhost:15672/

帐号密码guest

生产者

using RabbitMQ.Client;
using System.Text;

var conFactory = new ConnectionFactory();

conFactory.HostName = "localhost";
//conFactory.Port = 15672;
conFactory.DispatchConsumersAsync = true;
string exchangeName = "exchange1";
var connection = conFactory.CreateConnection();

while (true)
{
    using var channel = connection.CreateModel();
    var prop = channel.CreateBasicProperties();
    prop.DeliveryMode = 2;
    channel.ExchangeDeclare(exchangeName, "direct");
    byte[] bytes = Encoding.UTF8.GetBytes(DateTime.Now.ToString());
    channel.BasicPublish(exchangeName, routingKey: "key1", mandatory: true, basicProperties: prop, body: bytes);
    Console.WriteLine("Ok"+DateTime.UtcNow);
    Thread.Sleep(1000);
}

消费者

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

var connfactory = new ConnectionFactory();
connfactory.HostName = "localhost";
connfactory.DispatchConsumersAsync = true;
var connection = connfactory.CreateConnection();
string exchangeName = "exchange1";
string queueName = "queue1";
string routingkey = "key1";

using var channel = connection.CreateModel();

channel.ExchangeDeclare(exchangeName, "direct");
channel.QueueDeclare(queueName, durable: true,
        exclusive: false, autoDelete: false, arguments: null);
channel.QueueBind(queue: queueName,
    exchange: exchangeName, routingKey: routingkey);

AsyncEventingBasicConsumer consumer = new AsyncEventingBasicConsumer(channel);
consumer.Received += Consumer_Received;

channel.BasicConsume(queue: queueName, autoAck: false, consumer: consumer);
Console.WriteLine("按回车退出");
Console.ReadLine();
async Task Consumer_Received(object sender, BasicDeliverEventArgs args)
{
    try
    {
        var bytes = args.Body.ToArray();
        string msg = Encoding.UTF8.GetString(bytes);
        Console.WriteLine(DateTime.Now + "收到了消息" + msg);
        channel.BasicAck(args.DeliveryTag, multiple: false);
        await Task.Delay(800);
    }
    catch (Exception ex)
    {
        channel.BasicReject(args.DeliveryTag, true);
        Console.WriteLine("处理收到的消息出错" + ex);
    }
}