RabbitMQ处理MES调度任务

消息队列-简述

个人总结:消息队列是一个主要为了处理高并发业务场景的中间件，核心数据结构是队列，即先进先出，结合经典秒杀场景，将一个个秒杀请求存储在消息队列中，通过路由分发到一个个消费者异步处理消息，从而避免了，服务器在短时间内应对大量高并发请求而导致服务瘫痪的尴尬场景。
推荐一篇介绍得很好的博客介绍:消息队列技术介绍
我使用的消息队列是RabbitMQ，语言是C#

安装Erlang环境(是安装RabbitMQ服务需要的一个环境)

Windows下载地址:Erlang-Windows-Installer
安装好后，需要配置环境变量，配置完后，调出控制台，查看erl版本号:

至此，erlang安装成功

安装RabbitMQ服务(Rabbitmq-server)

rabbitmq-server官网地址:rabbitmq-server官网
在GetStarted下面找到下载链接，最后找到Windows安装包如下图下载地址:

下载安装略. rabbitMq官网上有很多关于rabbitMq的介绍，以及在相关语言环境下使用

RabbitMQ可视化后台管理界面

找到我们刚刚安装好的RabbitMq-server安装路径下:

C:\Program Files\RabbitMQ Server\rabbitmq_server-3.9.14\sbin

在这里打开cmd控制台，或者将此路径添加到环境变量中，输入以下命令(启动管理插件，可视化界面)：

rabbitmq-plugins enable rabbitmq_management

转-RabbitMq常用命令:RabbitMq常用命令
启动之后，在浏览器上打开可视化管理界面：
本地可视化管理界面地址：http://127.0.0.1:15672

默认用户名和密码都是：guest

基本消息发送

创建连接消息队列配置信息实体：

public class RabbitMqModel{/// <summary>/// 服务器ip地址/// </summary>private string host;public string Host{get { return host; }set { host = value; }}/// <summary>///服务器端口/// </summary>private int port;public int Port{get { return port; }set { port = value; }}/// <summary>/// 消息队列登录用户名/// </summary>private string userName;public string UserName{get { return userName; }set { userName = value; }}/// <summary>/// 消息队列登录密码/// </summary>private string passWord;public string PassWord{get { return passWord; }set { passWord = value; }}/// <summary>/// 消息队列名称/// </summary>private string queueName;public string QueueName{get { return queueName; }set { queueName = value; }}/// <summary>/// 交换机名称/// </summary>private string exchangeName;public string ExchangeName{get { return exchangeName; }set { exchangeName = value; }}}

初始化RabbitMqModel：

private static RabbitMqModel _rabbitMqModel;
_rabbitMqModel = new RabbitMqModel(){Host = "localhost",Port = 5672,UserName = "guest",PassWord = "guest",QueueName = "testQue1",ExchangeName = "testExchange"};
RabbitMQUtil.InitFactory(_rabbitMqModel);public static void InitFactory(RabbitMqModel mqModel){if (_connectionFactory != null) return;_connectionFactory = new ConnectionFactory(){HostName = mqModel.Host,Port = mqModel.Port,UserName = mqModel.UserName,Password = mqModel.PassWord};}

生产者（消息发送）：

public static bool SendMsgToServer(object content,string queueName,string exchangeName,ILog log){if (_connectionFactory == null){// 发送消息必须先实例化工厂return false;}bool flag;_connectionFactory.RequestedHeartbeat= TimeSpan.FromSeconds(10);try{using (var connection = _connectionFactory.CreateConnection()){//创建一个通道，这个就是Rabbit自己定义的规则了，如果自己写消息队列，这个就可以开脑洞设计了using (var channel = connection.CreateModel()){//声明交换机exchangchannel.ExchangeDeclare(exchange: "myexchange",type: ExchangeType.Direct,durable: false,autoDelete: false,arguments: null);channel.QueueDeclare(queueName, false, false, false,arguments: new Dictionary<string, object>() {{ "x-max-priority",10 }  //设置队列优先级，如果不设置消息优先级，则不生效，前提是必须先创建好一个带有"x-max-priority"属性的队列，否则会报错；});channel.QueueBind(queue: queueName, exchange: "myexchange", ""); // 绑定队列名称和exchange路由channel.ConfirmSelect();  //确认消息是否发送成功// 序列化对象 发送消息string message = JsonConvert.SerializeObject(content);var properties = channel.CreateBasicProperties();properties.Persistent = true;  //消息持久化，队列没有设置持久化，同样不会生效，队列持久化durable:trueproperties.Priority = 1; // 设置消息优先级var body = Encoding.UTF8.GetBytes(message);channel.BasicPublish("myexchange", "",properties, body); //发送消息var isOk = channel.WaitForConfirms();if (!isOk){//throw new Exception("The message is not reached to the server!");flag = false;}flag = true;}}}catch (Exception ex){log.Error(ex.Message, ex);flag = false;}return flag;}

消费者（接收消息）：

public class Consumer{public static IConnection _connection;public static IModel _channel;public static void ConsumerMessage(string queueName){try{if (_connection == null) return;_channel.BasicQos(0, 1, false); // 设置当没有收到上条消息处理完时，下一条消息将不被处理var consumer = new EventingBasicConsumer(_channel);consumer.Received += (model, ea) =>{string message = Encoding.UTF8.GetString(ea.Body.ToArray());//接收消息TaskInfoDTO taskInfo = JsonConvert.DeserializeObject<TaskInfoDTO>(message);// 模拟处理任务，延时5秒Thread.Sleep(5000);try{// 判断任务是否存在if (RabbitMQUtil._keys.Count > 0 && RabbitMQUtil._keys.Contains(taskInfo.Task_Id)){//  模拟数据库修改任务状态TaskInfoDTO task=null;foreach (var item in MainWindow.data.TaskInfos){if (item.Task_Id == taskInfo.Task_Id){task = item;break;}}if (task != null){Application.Current.Dispatcher.Invoke(() =>{MainWindow.data.TaskInfos.Remove(task);MainWindow.RefreshDataGrid();});_channel.BasicAck(ea.DeliveryTag, false); // 标志消息已被处理RabbitMQUtil._keys.Remove(taskInfo.Task_Id); // 删除缓存}else{//这里可以将消息重回队列，也可将消息删除，(可以设置一个删除队列，将无用的消息放到删除队列)}}}catch (Exception ex){LogUtils.log.Error(ex.Message, ex);}};_channel.BasicConsume(queueName, false, consumer);  // 设置手动交付消息}catch (Exception ex){LogUtils.log.Error(ex.Message, ex);}}}

设置消息优先级发送

设置带优先级的发送消息，生产者做部分修改，消费者不做改变：

public static bool SendMsgToServerWithLevel(object content, string queueName, string exchangeName,ILog log){if (_connectionFactory == null){// 发送消息必须先实例化工厂return false;}bool flag;_connectionFactory.RequestedHeartbeat = TimeSpan.FromSeconds(10);try{using (var connection = _connectionFactory.CreateConnection()){using (var channel = connection.CreateModel()){//声明交换机exchangchannel.ExchangeDeclare(exchange: "myexchange",type: ExchangeType.Direct,durable: false,autoDelete: false,arguments: null);channel.QueueDeclare(queueName, false, false, false, arguments:new Dictionary<string, object>() {{ "x-max-priority",10 }});channel.QueueBind(queue: queueName, exchange: "myexchange","");channel.ConfirmSelect();// 序列化对象 发送消息string message = JsonConvert.SerializeObject(content);var properties = channel.CreateBasicProperties();JObject obj = JsonConvert.DeserializeObject<JObject>(message);if (!string.IsNullOrEmpty(obj["Remark"].ToString()) && obj["Remark"].ToString().Equals("vip")){properties.Priority = 9;// 设置优先级}else{properties.Priority = 1;}properties.Persistent = true; // 设置消息持久化var body = Encoding.UTF8.GetBytes(message);channel.BasicPublish("myexchange", "", properties,body);var isOk = channel.WaitForConfirms();if (!isOk){//throw new Exception("The message is not reached to the server!");flag = false;}flag = true;}}}catch (Exception ex){log.Error(ex.Message, ex);flag = false;}return flag;}

消息发送确认机制

示例文章：发送消息确认

消息持久化、消息确认、优先级

示例文章：消息持久化、消息确认、优先级

不被重复消费

示例文章：消息不被重复消费

RabbitMQ各个参数含义

示例文章：rabbitMq各个参数含义
示例文章：重要方法简述

基础队列生产消费录屏

RabbitMq基本消费队列

优先级队列生产消费录屏

RabbitMq优先级消费者队列

代码上传

源代码：rabbitMq学习

以上示例文章均转载其他博主优秀文章，如有侵权请告知删除，谢谢。