C# RabbitMQ的使用

王林的关门女弟子 發表於 2021-12-28 21:01:00

本文目的如题。
<h1 id="安装">安装</h1>
先说一下RabbitMQ的安装，建议使用Docker镜像安装，Docker安装的好处是不管Windows系统还是Linux，安装步骤少，安装方法相同，不容易出错。使用下面的命令就可以：
<pre><code>docker run -d --hostname myRabbit --name rabbitmq3.9.11 -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin RABBITMQ_DEFAULT_VHOST=my_vhost -p 15672:15672 -p 5672:5672 rabbitmq3.9.11:management
</code></pre>
安装完成后，可以打开浏览器访问管理网站http://127.0.0.1:15672，使用安装时设置的用户名和密码登录，就可以进行管理了。
不管使用什么方法安装，都可以运行本文中的示例。这些示例中使用了用户admin，密码是admin，如果没有，可以在管理网站中创建： 
<img src="https://img2020.cnblogs.com/blog/4131/202112/4131-20211227102418672-757486571.png" alt="" loading="lazy"> 
本文的示例中还使用了my_vhost虚拟主机，如果没有，也需要定义一下： 
<img src="https://img2020.cnblogs.com/blog/4131/202112/4131-20211227102545114-68729408.png" alt="" loading="lazy"> 
注意，admin 需要有对my_vhost的操作权限。
<h1 id="编写消息接收端">编写消息接收端</h1>
<img src="https://img2020.cnblogs.com/blog/4131/202112/4131-20211227120536634-1910201730.png" alt="" loading="lazy">
安装完成后可以进行开发了。我们需要编写消息的生产者和消费者，如果哪一部分出了问题，或者RabbitMQ服务器出了问题，都会影响工作的进展。因此我们分步进行，先编写消息接受部分，也就是所谓的消费者，与RabbitMQ服务器联调，成功后再进行下一步。
先创建一个.Net 6的控制台项目,可以使用Visual Studio创建。如果使用命令行，命令如下:
<pre><code>mkdir DirectReceiveDemo
cd DirectReceiveDemo
dotnet new console
</code></pre>
然后安装rabbitmq.client程序包：
<pre><code>dotnet add package rabbitmq.client
</code></pre>
编写Program.cs代码如下：
<pre><code>using RabbitMQ.Client;
using System.Text;
using RabbitMQ.Client.Events;

var factory = new ConnectionFactory()
{
HostName = "127.0.0.1",
UserName = "admin",
Password = "admin",
VirtualHost = "my_vhost"
};
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
channel.QueueDeclare(queue: "mymessage",
 durable: false,
 exclusive: false,
 autoDelete: false,
 arguments: null);
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
 var body = ea.Body;
 var message = Encoding.UTF8.GetString(body.ToArray());
 Console.WriteLine("收到消息 {0}", message);
};
channel.BasicConsume(queue: "mymessage",
 autoAck: true,
 consumer: consumer);

Console.WriteLine(" 按回车退出");
Console.ReadLine();
}

</code></pre>
执行dotnet run 运行代码，程序会一直等待输入，这时需要输入一些消息验证程序。现在登录管理网站http://127.0.0.1:15672/，使用安装时设置的用户名和密码,在Connections分页中可以看到多了新的连接：
<img src="https://img2020.cnblogs.com/blog/4131/202112/4131-20211225161420761-896750400.png" alt="" loading="lazy"> 
在Channel分页中可以看到当前的Chanel: 
<img src="https://img2020.cnblogs.com/blog/4131/202112/4131-20211225163240422-1898796449.png" alt="" loading="lazy"> 
进入Queues分页，点击列表中的mymessage 
<img src="https://img2020.cnblogs.com/blog/4131/202112/4131-20211225163539307-455134765.png" alt="" loading="lazy"> 
进入mymessage队列： 
<img src="https://img2020.cnblogs.com/blog/4131/202112/4131-20211225163649775-1025754802.png" alt="" loading="lazy">
在Publish message中写一些消息并发送。回到控制台接收程序，消息应该已经被接收了。 
<img src="https://img2020.cnblogs.com/blog/4131/202112/4131-20211225163907252-167949991.png" alt="" loading="lazy">
到这里，接收部分完成，退出这个程序，我们开始编写发送部分。
<h1 id="编写发送端">编写发送端</h1>
<img src="https://img2020.cnblogs.com/blog/4131/202112/4131-20211227120603653-1746363739.png" alt="" loading="lazy">
创建过程跟接收部分完全一样，只是项目名称为DirectSendDemo，Program.cs代码如下：
<pre><code>using RabbitMQ.Client;
using System.Text;

var factory = new ConnectionFactory()
{
HostName = "127.0.0.1",
UserName = "admin",
Password = "admin",
VirtualHost = "my_vhost"
};
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
channel.QueueDeclare(queue: "mymessage",
 durable: false,
 exclusive: false,
 autoDelete: false,
 arguments: null);

Console.WriteLine("输入需要传输的消息，输入Exit退出");
var message = Console.ReadLine();
while (message != "Exit")
{
 var body = Encoding.UTF8.GetBytes(message);

 channel.BasicPublish(exchange: "",
 routingKey: "mymessage",
 basicProperties: null,
 body: body);
 Console.WriteLine(" 发送消息 {0}", message);
 message = Console.ReadLine();
}
}

Console.WriteLine("按回车退出");
Console.ReadLine();
</code></pre>
运行这个项目，输入一些消息， 
<img src="https://img2020.cnblogs.com/blog/4131/202112/4131-20211225164644357-455832030.png" alt="" loading="lazy"> 
还是回到管理页面，在mymessage队列页面，执行GetMessage,可以获取发送的消息。 
<img src="https://img2020.cnblogs.com/blog/4131/202112/4131-20211225164804085-577520464.png" alt="" loading="lazy">
<h1 id="测试发送端和接收端">测试发送端和接收端</h1>
<img src="https://img2020.cnblogs.com/blog/4131/202112/4131-20211227120436585-956069690.png" alt="" loading="lazy"> 
现在我们可以让发送和接收一起工作了，在两个终端分别启动发送和接收程序，看是否可以一起工作。 
<img src="https://img2020.cnblogs.com/blog/4131/202112/4131-20211225165733763-388671676.png" alt="" loading="lazy"> 
发送和接收可以一起工作了。
现在可以用这两个程序做一些测试，首先看一下一个发送端，两个接收端是什么情况： 
<img src="https://img2020.cnblogs.com/blog/4131/202112/4131-20211225170207922-402926912.png" alt="" loading="lazy"> 
我们发现，接收端会轮流接收消息。 
两个发送端对一个接收端的情况如下： 
<img src="https://img2020.cnblogs.com/blog/4131/202112/4131-20211225170450923-275660073.png" alt="" loading="lazy"> 
跟想象的一样，接收端会处理所有消息。
<h1 id="fanout-模式">Fanout 模式</h1>
<img src="https://img2020.cnblogs.com/blog/4131/202112/4131-20211227121051961-1631035887.png" alt="" loading="lazy">
现在我们需要处理一个消息有多个消费者的情况，这种情况下，消息需要发送给交换机（exchange）,然后将交换机与消息队列绑定，一个交换机可以绑定多个消息队列，这样，不同的消息消费者都可以接收到消息。我们创建一个新的发送方FanoutSender，将消息发送给exchange：
<pre><code>using RabbitMQ.Client;
using System.Text;

var factory = new ConnectionFactory()
{
HostName = "127.0.0.1",
UserName = "admin",
Password = "admin",
VirtualHost = "my_vhost"
};
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
channel.ExchangeDeclare("example.exchange", ExchangeType.Fanout, true, false, null);

Console.WriteLine("输入需要传输的消息，输入Exit退出");
var message = Console.ReadLine();
while (message != "Exit")
{
 var body = Encoding.UTF8.GetBytes(message);

 channel.BasicPublish(exchange: "example.exchange",
 routingKey: "",
 basicProperties: null,
 body: body);
 Console.WriteLine(" 发送消息 {0}", message);
 message = Console.ReadLine();
}
}

Console.WriteLine("按回车退出");
Console.ReadLine();
</code></pre>
然后创建两个接收方，FanoutReceiver1和FanoutReceiver2，分别接收que1和que2队列的消息，这两个队列都绑定到相同的交换机，代码如下： 
FanoutReceiver1:
<pre><code>using RabbitMQ.Client;
using System.Text;
using RabbitMQ.Client.Events;

var factory = new ConnectionFactory()
{
HostName = "127.0.0.1",
UserName = "admin",
Password = "admin",
VirtualHost = "my_vhost"
};
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
channel.ExchangeDeclare(exchange: "example.exchange",
type: "fanout", durable: true);

channel.QueueDeclare(queue: "que1",
 durable: true,
 exclusive: false,
 autoDelete: false,
 arguments: null);
channel.QueueBind(queue: "que1", exchange: "example.exchange",
routingKey: "");
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
 var body = ea.Body;
 var message = Encoding.UTF8.GetString(body.ToArray());
 Console.WriteLine("收到消息 {0}", message);
};
channel.BasicConsume(queue: "que1",
 autoAck: true,
 consumer: consumer);

Console.WriteLine(" 按回车退出");
Console.ReadLine();
}

</code></pre>
FanoutReceiver2:
<pre><code>using RabbitMQ.Client;
using System.Text;
using RabbitMQ.Client.Events;

var factory = new ConnectionFactory()
{
HostName = "127.0.0.1",
UserName = "admin",
Password = "admin",
VirtualHost = "my_vhost"
};
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
channel.ExchangeDeclare(exchange: "example.exchange",
type: "fanout", durable: true);

channel.QueueDeclare(queue: "que2",
 durable: true,
 exclusive: false,
 autoDelete: false,
 arguments: null);
channel.QueueBind(queue: "que2", exchange: "example.exchange",
routingKey: "");
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
 var body = ea.Body;
 var message = Encoding.UTF8.GetString(body.ToArray());
 Console.WriteLine("收到消息 {0}", message);
};
channel.BasicConsume(queue: "que2",
 autoAck: true,
 consumer: consumer);

Console.WriteLine(" 按回车退出");
Console.ReadLine();
}
</code></pre>
同时启动这三个程序，运行结果如下： 
<img src="https://img2020.cnblogs.com/blog/4131/202112/4131-20211225204608952-1931022465.png" alt="" loading="lazy"> 
发送的消息被同时接收。
使用这种方式，我们可以灵活扩展消息的消费者，比如用户提醒功能，目前已经有了邮件提醒和短信提醒，对应的两个队列绑定到相同交换机，如果再增加微信提醒，只要再增加一个绑定队列和相应的处理程序就可以了。
<h1 id="direct模式和routekey">Direct模式和RouteKey</h1>
<img src="https://img2020.cnblogs.com/blog/4131/202112/4131-20211227151128469-625485335.png" alt="" loading="lazy"> 
在Fanout模式下，我们将消息发送到订阅消息的所有队列中，如果我们希望选择性地向队列发送消息，可以使用Direct模式，根据不同的RouteKey向不同的队列发送消息。
我们建立三个控制台程序程序模拟一个发送方和两个接收方，项目的创建方法同上，代码如下： 
发送：
<pre><code>using RabbitMQ.Client;
using System.Text;

var factory = new ConnectionFactory()
{
HostName = "127.0.0.1",
UserName = "admin",
Password = "admin",
VirtualHost = "my_vhost"
};
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
channel.ExchangeDeclare("directdemo.exchange", ExchangeType.Direct, true, false, null);

Console.WriteLine("输入需要传输的消息，输入Exit退出");
var message = Console.ReadLine();
while (message != "Exit")
{
 Console.WriteLine("输入RouteKey");
 var routekey = Console.ReadLine();
 var body = Encoding.UTF8.GetBytes(message);

 channel.BasicPublish(exchange: "directdemo.exchange",
 routingKey: routekey,
 basicProperties: null,
 body: body);
 Console.WriteLine(" 发送消息 {0} Routekey {1}", message,routekey);
 message = Console.ReadLine();
}
}

Console.WriteLine("按回车退出");
Console.ReadLine();
</code></pre>
接收1：
<pre><code>using RabbitMQ.Client;
using System.Text;
using RabbitMQ.Client.Events;

var factory = new ConnectionFactory()
{
HostName = "127.0.0.1",
UserName = "admin",
Password = "admin",
VirtualHost = "my_vhost"
};
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{

channel.ExchangeDeclare(exchange: "directdemo.exchange",
type: ExchangeType.Direct, durable: true);

channel.QueueDeclare(queue: "log_que",
 durable: true,
 exclusive: false,
 autoDelete: false,
 arguments: null);
channel.QueueBind(queue: "log_que", exchange: "directdemo.exchange",
routingKey: "log");
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
 var body = ea.Body;
 var message = Encoding.UTF8.GetString(body.ToArray());
 Console.WriteLine("收到消息 {0}", message);
};
channel.BasicConsume(queue: "log_que",
 autoAck: true,
 consumer: consumer);

Console.WriteLine(" 按回车退出");
Console.ReadLine();
}
</code></pre>
接收2：
<pre><code>using RabbitMQ.Client;
using System.Text;
using RabbitMQ.Client.Events;

var factory = new ConnectionFactory()
{
HostName = "127.0.0.1",
UserName = "admin",
Password = "admin",
VirtualHost = "my_vhost"
};
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{

channel.ExchangeDeclare(exchange: "directdemo.exchange",
type: ExchangeType.Direct, durable: true);

channel.QueueDeclare(queue: "email_que",
 durable: true,
 exclusive: false,
 autoDelete: false,
 arguments: null);
channel.QueueBind(queue: "email_que", exchange: "directdemo.exchange",
routingKey: "email");
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
 var body = ea.Body;
 var message = Encoding.UTF8.GetString(body.ToArray());
 Console.WriteLine("收到消息 {0}", message);
};
channel.BasicConsume(queue: "email_que",
 autoAck: true,
 consumer: consumer);

Console.WriteLine(" 按回车退出");
Console.ReadLine();
}

</code></pre>
上面的代码中，关键是队列绑定：
<pre><code> channel.QueueBind(queue: "email_que", exchange: "directdemo.exchange",
routingKey: "email");
</code></pre>
这句话将queue、exchange和routingKey绑定在一起。运行效果如下： 
<img src="https://img2020.cnblogs.com/blog/4131/202112/4131-20211227161524016-361427695.png" alt="" loading="lazy">
<h1 id="topic-模式">Topic 模式</h1>
<img src="https://img2020.cnblogs.com/blog/4131/202112/4131-20211227161744880-708206835.png" alt="" loading="lazy">
前面的Direct模式中，RouteKey是固定的，Topic模式引入了通配符，RouteKey可以是符合表达式的任何字符串。
<ul>
<li>通配符“*”，代表一个字符</li>
<li>通配符“#”,代表0或多个字符</li>
</ul>
仔细研究上面的规则，会发现Topic模式可以代替Direct和Fanout，如果RouteKey被设置为“#”，就是队列可以接收任何消息，这与Fanout模式相同，如果RouteKey中没有通配符，则和使用Direct模式的效果相同。
现在我们编写Topic模式的发送和接收，代码如下： 
Topic模式发送：
<pre><code>using RabbitMQ.Client;
using System.Text;

var factory = new ConnectionFactory()
{
HostName = "127.0.0.1",
UserName = "admin",
Password = "admin",
VirtualHost = "my_vhost"
};
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
channel.ExchangeDeclare("topicdemo.exchange", ExchangeType.Topic, true, false, null);

Console.WriteLine("输入需要传输的消息，输入Exit退出");
var message = Console.ReadLine();
while (message != "Exit")
{
 Console.WriteLine("输入RouteKey");
 var routekey = Console.ReadLine();
 var body = Encoding.UTF8.GetBytes(message);

 channel.BasicPublish(exchange: "topicdemo.exchange",
 routingKey: routekey,
 basicProperties: null,
 body: body);
 Console.WriteLine(" 发送消息 {0} Routekey {1}", message, routekey);
 message = Console.ReadLine();
}
}

Console.WriteLine("按回车退出");
Console.ReadLine();
</code></pre>
Topic模式接收：
<pre><code>using RabbitMQ.Client;
using System.Text;
using RabbitMQ.Client.Events;

var factory = new ConnectionFactory()
{
HostName = "127.0.0.1",
UserName = "admin",
Password = "admin",
VirtualHost = "my_vhost"
};
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{

channel.ExchangeDeclare(exchange: "topicdemo.exchange",
type: ExchangeType.Topic, durable: true);

channel.QueueDeclare(queue: "topic_que",
 durable: true,
 exclusive: false,
 autoDelete: false,
 arguments: null);
channel.QueueBind(queue: "topic_que", exchange: "topicdemo.exchange",
routingKey: "#.log");
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
 var body = ea.Body;
 var message = Encoding.UTF8.GetString(body.ToArray());
 Console.WriteLine("收到消息 {0}", message);
};
channel.BasicConsume(queue: "topic_que",
 autoAck: true,
 consumer: consumer);

Console.WriteLine(" 按回车退出");
Console.ReadLine();
}

</code></pre>
我们设置的RouteKey是"#.log"，也就是匹配这个表达式的RouteKey的消息会被接收到： 
<img src="https://img2020.cnblogs.com/blog/4131/202112/4131-20211227163714700-69756575.png" alt="" loading="lazy">
到这里RabbitMQ常用的几种模式都介绍了，最后说一点代码中的细节，在发送方和接收方代码中，有重复的queue或者exchange声明，比如:
<pre><code> channel.QueueDeclare(queue: "mymessage",
 durable: false,
 exclusive: false,
 autoDelete: false,
 arguments: null);
</code></pre>
这些代码让人感到有些困惑，似乎每次都需要声明，而实际上是只要存在相关的queue或者exchange，这些代码就不再起作用。之所以在发送方和接收方都包含这些代码，是因为不知道是否存在相关的queue或exchange,也不知道谁先启动，避免出错。如果在RabbitMQ的Web管理页面预先手工创建了相应的queue或者exchange，这些代码是可以去掉的。
本文代码可以从github下载：https://github.com/zhenl/ZL.RabbitMQ.Demo

</div>
<div id="MySignature" role="contentinfo">
本文来自博客园，作者：寻找无名的特质，转载请注明原文链接：https://www.cnblogs.com/zhenl/p/15730784.html 
来源：https://www.cnblogs.com/zhenl/p/15730784.html

頁: [1]

琼殿技术论坛's Archiver

C# RabbitMQ的使用