消息队列收发消息Demo

空蝉如此世 · 发表于 2019-5-30 22:58:30

前言

项目中，有些地方为了保证信息处理的准确性，会使用到消息队列。比如：高并发情况下，网站进行数据处理。今天本文记录RabbitMQ简单的一个demo。在进入本文之前，RbMQ环境必须提前安装好（因为我本机RbMQ环境之前安装好了，怕卸载造成一些注册表清不干净的问题，这里就不记录环境的安装）。

环境&工具

>Windows 10 系统
>RabbitMQ环境服务
>VS 2017

正文

1：新建两个控制台应用程序，分别为 “RabbitMQClient” 和 “RabbitMQServer”

2：引用 “RabbitMQ.Client” dll

3：在RabbitMQClient 项目，Program.cs中，贴入以下代码

using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Text;
namespace RabbitMQClientt
{
class Program
{
private static readonly ConnectionFactory rabbitMqFactory = new ConnectionFactory()
{
HostName = "localhost",
UserName = "guest",
Password = "guest",
Port = 5672
};
static void Main(string[] args)
{
Worker();
}
private static void Worker()
{
string exchangeName = "SendEmail.exchange";
const string queueName = "SendEmail.queue";
using (IConnection conn = rabbitMqFactory.CreateConnection())
{
using (IModel channel = conn.CreateModel())
{
//2 定义一个exchange
channel.ExchangeDeclare(exchangeName, "direct", durable: true, autoDelete: false, arguments: null);
//4 定义两个queue
channel.QueueDeclare(queueName, durable: true, exclusive: false, autoDelete: false, arguments: null);
channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);
Console.WriteLine(" [*] Waiting for messages.");
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
var msgBody = Encoding.UTF8.GetString(ea.Body);
Console.WriteLine(string.Format("接收时间:{0}，消息内容：{1}", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"), msgBody));
channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
};
channel.BasicConsume(queueName, autoAck:false, consumer: consumer);
Console.WriteLine("按任意值，退出程序");
Console.ReadKey();
}
}
}
}
}

复制代码

4：在RabbitMQServer项目，Program.cs中，贴入以下代码

using RabbitMQ.Client;
using System;
using System.Text;
namespace RabbitMQServer
{
class Program
{
private static readonly ConnectionFactory rabbitMqFactory = new ConnectionFactory()
{
HostName = "localhost",
UserName = "guest",
Password = "guest",
Port = 5672
};
static void Main(string[] args)
{
CommonSample();
}
public static void SendMessage(IModel channel,IBasicProperties props,String exchangeName, String queueName, String message)
{
var msgBody = Encoding.UTF8.GetBytes(message);
channel.BasicPublish(exchange: exchangeName, routingKey: queueName, basicProperties: props, body: msgBody);
Console.WriteLine(string.Format("发送完成，发送时间:{0}", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")));
}
private static void CommonSample()
{
string exchangeName = "SendEmail.exchange";
const string queueName = "SendEmail.queue";
using (IConnection conn = rabbitMqFactory.CreateConnection())
using (IModel channel = conn.CreateModel())
{
//2 定义一个exchange
channel.ExchangeDeclare(exchangeName, "direct", durable: true, autoDelete: false, arguments: null);
//4 定义两个queue
channel.QueueDeclare(queueName, durable: true, exclusive: false, autoDelete: false, arguments: null);
//3 定义exchange到queue的binding
channel.QueueBind(queueName, exchangeName, routingKey: queueName);
var props = channel.CreateBasicProperties();
props.Persistent = true;
ConsoleKeyInfo kinfo;
StringBuilder result = new StringBuilder();
do
{
kinfo = Console.ReadKey();
if (kinfo.Key == ConsoleKey.Enter)
{
String content = result.ToString();
Console.WriteLine();
Console.WriteLine($"您输入的是:{content}");
SendMessage(channel, props, exchangeName, queueName, content);
result.Clear();
}
else
{
result.Append(kinfo.KeyChar);
}
} while (kinfo.Key != ConsoleKey.Escape);
}
}
}
}

复制代码

5：F5 启动 RabbitMQServer 项目，在控制台中，输入信息，回车。

6：不要关闭 RabbitMQServer 项目控制台，右击 RabbitMQClient 项目——> “调试”——>“启动新实例”

可在右边 RabbitMQServer 控制台里面输入信息，即可显示在左边。RabbitMQServer 项目进行消息的生产。RabbitMQClient项目则对消息进行处理(消费)。这就是消息队列中的生产者和消费者。代码里还涉及到路由，信道。

在消息生产者这边，把要处理的信息丢到消息队列中。消费者这边在接收到消息后，可对这条数据进行相应的处理。

环境安装好的话，本地默认的端口地址应该是 http://localhost:15672/#/queues。直接 guest 账户访问，也可自己敲命令行配置账户和相应的权限。

刚才在RabbitMQServer 控制台输入的信息，其实都先存在消息队列中，可先将 RabbitMQClient 窗口关闭，再在窗口输入几条信息，观察消息队列，可看到队列中有三条数据。

当我们RabbitMQClient 窗口启动，消息在正常接收后，会自动在消息队列中消失。这个是消息队列的确认机制，一旦准确接收到了我们需要的信息，消费者这边会向消息队列返回一个状态，用来销毁这条消息。如果消费者没有接收到消息，或者出现异常，没有返回这个状态值，那么这条消息将会一直在队列中，直到消息被正常的消费，这样确保了消息的准确性。

尾声

很遗憾，在项目中我用到消息队列的地方很少。没有进行深入的了解，只会基础的使用。

上面demo 里面有一个坑，在RabbitMQClient 里面 RbMQ 的连接套了一层 using。因为是异步线程，这样会造成程序跑的时候，还没接收完消息，就把RbMQ Connection 给回收了，会报错。处理方法是：去掉using,手动释放。

小渣渣 · 发表于 2019-5-31 09:12:21

最好把文章的 demo 源码，上传到：https://down.itsvse.com/

		自动登录	找回密码
密码			注册[Register]

[插件库] 消息队列收发消息Demo

相关资源下载