RabbitMQ安装以及简单操作应用(针对Windows和C#)

2023-07-29,,

1.RabbitMQ安装

1.1下载并安装Erlang

https://www.erlang.org/downloads

一直点next就安装好了。我直接使用了默认的安装目录。否则的话,应该需要配置一下环境变量吧。

1.2下载并安装RabbitMQ

https://www.rabbitmq.com/download.html

也是一直点next,同样使用了默认的安装目录。

2.配置RabbitMQ

2.1激活RabbitMQ's Management Plugin

打开命令提示符,按下图输入

这一步用不用管理员权限都行。但是重启RabbitMQ需要有管理员权限。

2.2查看、创建用户,角色授权

查看已有用户及角色

rabbitmqctl.bat list_users

创建新用户

rabbitmqctl.bat add_user userName password

给新用户授权administrator角色

rabbitmqctl.bat set_user_tags userName administrator

删除某个用户

rabbitmqctl.bat delete_user userName

更改密码

rabbitmqctl.bat change_password userName newPassword

2.3登录RabbitMQ

打开http://localhost:15672,进入RabbitMQ管理页面,使用用户名和密码登录系统。

3.RabbitMQ简单操作

3.1消息产生者的控制台应用

新建一个名为EventProducer的控制台项目,添加RabbitMQ.Client包。(我这里用的是VS2019创建的.NET CORE的控制台项目)

添加NuGet包,可以通过菜单栏上的“工具->NuGet包管理器->程序包管理控制台”,然后选择正确的“默认项目”,输入Install-Package RabbitMQ.Client的方式进行安装,也可以通过在右侧的“解决方案资源管理器”上右键该项目的依赖项,选择“管理NuGet程序包”,在打开的页面中,输入“RabbitMQ.Client”,然后选择安装。

EventProducer项目发送事件到RabbitMQ Server上。首先要通过ConnectionFactory使用CreateConnection方法与RabbitMQ Server建立连接。

var factory = new ConnectionFactory()
{
HostName = "localhost",
UserName = "larissa",
Password = "hzbp"
}; using(var connection = factory.CreateConnection())

每个connection只连接一个服务器。该连接是基于socket实现的。RabbitMQ基于AMQP,AMQP使用TCP协议。连接服务器后,使用IConnection的CreateModel()方法创建消息通道(channel)。在这里就把model这个单词理解为channel吧。一个connection中可以包含多条消息通道。

var channel = connection.CreateModel();

消息通道创建好了之后,需要声明交换机(exchange)和队列(queue)。在RabbitMQ中,消息产生者、交换机、队列和消息消费者的关系如下图所示。

消息产生者(P)发布消息到交换机(X)上,交换机指定消息按照给定的规则,路由到给定的队列(Q)上。队列是消息载体,也就是说每条消息都是存放在一个或多个队列中的。然后广播到给定的消息消费者(C)。消息通道声明交换机的命令如下:

IModel.ExchangeDeclare(string exchange, string type, bool durable, bool autoDelete, IDictionary<string, object>arguments)

exchange: 交换机的名称

type:交换机的类型。共有四种类型:direct, fanout, topic。

direct,处理路由键(每条消息都有一个路由键,可以是事件名eventName),对消息路径进行全文匹配。
fanout,不处理路由键,将消息广播到绑定该交换机的所有队列上。
topic,对路由键进行模糊匹配。例如“#”表示一个或多个单词,“*”仅匹配一个单词。
headers,匹配AMQP消息的header而不是路由键,这个很少用到。

durable:持久化。如果是true的话,则服务器重启时不会丢失消息。

autoDelete:自动删除。自删除的前提是至少有一个队列或者交换器与该交换器绑定。一般都设置为false。

arguments:我一直没用到过这个参数。所以也没研究。

下面声明了一个交换机:

channel.ExchangeDeclare("RabbitMQ_Demo_Exchange", "direct", durable: true, autoDelete: false, arguments: null);

声明了交换机之后,消息通道还需要声明队列。

IModel.QueueDeclare(string queue, bool durable, bool exclusive, bool autoDelete, IDictionary<string, object>arguments)

与交换机声明相比,多了一个exclusive参数。exclusive为true的时候意味着队列是排他的,连接断开时将自动删除。所以一般我都是用的false。

下面声明了一个队列:

channel.QueueDeclare("UserQueue", durable:true, exclusive:false, autoDelete:false, arguments:null);

发布消息时,需要将该消息的路由键和交换机和队列绑定,通过IModel.QueueBind实现,如下:

IModel.QueueBind(string queue, string exchange, string routingKey, IDictionary<string, object>arguments)

在EventProducer项目中,我们发送一个AddUserEvent到消息队列中。AddUserEvent.cs的代码如下:

public class AddUserEvent
{
public string Name { get; }
public int Age { get; }
public AddUserEvent(string Name, int Age)
{
this.Name = Name;
this.Age = Age;
}
}

AddUserEvent

Program.cs的代码如下:

using System;
using RabbitMQ.Client;
using System.Text;
using Newtonsoft.Json; namespace EventProducer
{
class Program
{
const string ExchangeName = "RabbitMQ_Demo_Exchange";
const string QueueName = "UserQueue";
const string eventName = "AddUserEvent";
static void Main(string[] args)
{
var factory = new ConnectionFactory()
{
HostName = "localhost",
UserName = "larissa",
Password = "xxx"
}; using(var connection = factory.CreateConnection())
using(var channel = connection.CreateModel())
{
channel.ExchangeDeclare(ExchangeName, "direct", durable: true, autoDelete: false, arguments: null);
channel.QueueDeclare(QueueName, durable: true, exclusive:false, autoDelete: false, arguments: null);
channel.QueueBind(QueueName, ExchangeName, eventName, arguments: null); AddUserEvent _event = new AddUserEvent("larissa", 100);
var message = JsonConvert.SerializeObject(_event);
var body = Encoding.UTF8.GetBytes(message); channel.BasicPublish(
exchange: ExchangeName,
routingKey: eventName,
mandatory: true,
basicProperties: null,
body: body
); string _message = string.Format(DateTime.Now + "Sent {0} with message : {1}.", eventName, message);
Console.WriteLine(_message);
Console.Read();
}
}
}
}

Program.cs

运行后,可以在RabbitMQ的管理页面查看到刚才所声明的交换机和队列,以及它们之间的绑定。如下图所示。可见,已经成功了产生并发送了AddUserEvent事件到RabbitMQ的UserQueue队列中。

 3.2消息消费者的控制台应用

新建一个EventConsumer的控制台项目,并添加RabbitMQ.Client包。同样的,在消息消费者端也需要连接RabbitMQ Server、创建消息通道和声明交换机和队列。不然消费者怎么知道该接收哪个队列的消息呢。在本项目中,消费者端接收到一个AddUserEvent事件,并对其进行处理。因此,在消费者端也有一个一样的AddUserEvent.cs。Program.cs的代码如下:(这里用的是异步方法)

using System;
using System.Text;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Threading.Tasks; namespace EventConsumer
{
class Program
{
const string ExchangeName = "RabbitMQ_Demo_Exchange";
const string QueueName = "UserQueue";
static void Main(string[] args)
{
var factory = new ConnectionFactory()
{
HostName = "localhost",
UserName = "larissa",
Password = "xxx",
DispatchConsumersAsync=true // 注意这里,在异步处理事件的时候,这个一定要设置为true
}; using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
channel.ExchangeDeclare(ExchangeName, "direct", durable: true, autoDelete: false, arguments: null);
channel.QueueDeclare(QueueName, durable: true, exclusive: false, autoDelete: false, arguments: null); var consumer = new AsyncEventingBasicConsumer(channel);
consumer.Received += Consumer_Received; channel.BasicConsume(queue: QueueName, autoAck: true, consumer);
Console.Read();
}
} static async Task Consumer_Received(object sender, BasicDeliverEventArgs eventArgs)
{
var eventName = eventArgs.RoutingKey;
var message = Encoding.UTF8.GetString(eventArgs.Body.ToArray());
await ProcessEvent(eventName, message); } static async Task ProcessEvent(string eventName, string message)
{
await Task.Delay(1000);
string _message = string.Format("Receive {0} with {1}", eventName, message);
Console.WriteLine(_message);
}
}
}

Program.cs

其中,在下面这句中,autoAck为true时,意味着消息发送到操作系统的套接字缓冲区时,就表示任务消息已经被消费了。但如果此时套接字缓冲区崩溃,消息在未被消费者应用程序消费的情况下就被队列删除。如果想要保证消息可靠的达到消费者端,将autoAck设置为false,当套接字缓冲区崩溃时,仍能保证消息被重新消费。

channel.BasicConsume(queue: QueueName, autoAck: true, consumer);

运行结果如下图所示,显示已经从声明的队列中接收到AddUserEvent事件了。

主要参考:

https://www.cnblogs.com/ericli-ericli/p/5924608.html

和一些博客园的文章,忘记随手记录了。

RabbitMQ安装以及简单操作应用(针对Windows和C#)的相关教程结束。

《RabbitMQ安装以及简单操作应用(针对Windows和C#).doc》

下载本文的Word格式文档,以方便收藏与打印。