RabbitMQ .NET

setup rabbitmq

docker run --name=rabbit -p 15672:15672 -p 5672:5672 -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=xxx -d rabbitmq:management

ip:15672

login

nuget RabbitMQ.Client

Send

//1.1.实例化连接工厂
var factory = new ConnectionFactory() { HostName = "localhost" };
//2. 建立连接
using (var connection = factory.CreateConnection())
{
    //3. 创建信道
    using (var channel = connection.CreateModel())
    {
        //4. 申明队列
        channel.QueueDeclare(queue: "hello", durable: false, exclusive: false, autoDelete: false, arguments: null);
        //5. 构建byte消息数据包
        string message = args.Length > 0 ? args[0] : "Hello RabbitMQ!";
        var body = Encoding.UTF8.GetBytes(message);
        //6. 发送数据包
        channel.BasicPublish(exchange: "", routingKey: "hello", basicProperties: null, body: body);
        Console.WriteLine(" [x] Sent {0}", message);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

Received

//1.实例化连接工厂
var factory = new ConnectionFactory() { HostName = "localhost" };
//2. 建立连接
using (var connection = factory.CreateConnection())
{
    //3. 创建信道
    using (var channel = connection.CreateModel())
    {
        //4. 申明队列
        channel.QueueDeclare(queue: "hello", durable: false, exclusive: false, autoDelete: false, arguments: null);
        //5. 构造消费者实例
        var consumer = new EventingBasicConsumer(channel);
        //6. 绑定消息接收后的事件委托
        consumer.Received += (model, ea) =>
        {
            var message = Encoding.UTF8.GetString(ea.Body.ToArray());
            Console.WriteLine(" [x] Received {0}", message);
            Thread.Sleep(6000);//模拟耗时
            Console.WriteLine(" [x] Done");
        };
        //7. 启动消费者
        channel.BasicConsume(queue: "hello", autoAck: true, consumer: consumer);
        Console.WriteLine(" Press [enter] to exit.");
        Console.ReadLine();
    }
}


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

Err
RabbitMQ.Client.Exceptions.BrokerUnreachableException:“None of the specified endpoints were reachable”
IP host

Err
AuthenticationFailureException: ACCESS_REFUSED - Login was refused using authentication mechanism

new ConnectionFactory() { HostName = "localhost" ,UserName="admin",Password="xxx"}

1
2

消息持久化

		//...
      //4. 申明队列(指定durable:true,告知rabbitmq对消息进行持久化)
      channel.QueueDeclare(queue: "hellooo", durable: true, exclusive: false, autoDelete: false, arguments: null);
      //将消息标记为持久性 - 将IBasicProperties.SetPersistent设置为true
      var properties = channel.CreateBasicProperties();
      properties.Persistent = true;
      //5. 构建byte消息数据包
      string message = args.Length > 0 ? args[0] : "Hello RabbitMQ!";
      var body = Encoding.UTF8.GetBytes(message);
      //6. 发送数据包(指定basicProperties)
      channel.BasicPublish(exchange: "", routingKey: "hello", basicProperties: properties, body: body);
      //6. 发送数据包
      channel.BasicPublish(exchange: "", routingKey: "hello", basicProperties: null, body: body);
      Console.WriteLine(" [x] Sent {0}", message);
1
2
3
4
5
6
7
8
9
10
11
12
13
14

durable: true