运用.net core中实例讲解RabbitMQ_实用技巧_

 /// 
 /// RabbitMQ帮助类 /// 
 public class RabbitMQHelper { private static ConnectionFactory factory; private static object lockObj = new object(); /// 
 /// 获取单个RabbitMQ连接 /// 
 ///  public static IConnection GetConnection() { if (factory == null) { lock (lockObj) { if (factory == null) { factory = new ConnectionFactory { HostName = "172.16.2.84",//ip Port = 5672,//端口 UserName = "admin",//账号 Password = "123456",//密码 VirtualHost = "develop" //虚拟主机 }; } } } return factory.CreateConnection(); } }

 public static void SimpleSendMsg() { string queueName = "simple_order";//队列名 //创建连接 using (var connection = RabbitMQHelper.GetConnection()) { //创建信道 using (var channel = connection.CreateModel()) {//创建队列 channel.QueueDeclare(queueName, durable: false, exclusive: false, autoDelete: false, arguments: null); for (var i = 0; i < 10; i++) { string message = $"Hello RabbitMQ MessageHello,{i + 1}"; var body = Encoding.UTF8.GetBytes(message);//发送消息 channel.BasicPublish(exchange: "", routingKey: queueName, mandatory: false, basicProperties: null, body); Console.WriteLine($"发送消息到队列:{queueName},内容:{message}"); } } } }

 public static void SimpleConsumer() { string queueName = "simple_order"; var connection = RabbitMQHelper.GetConnection(); { //创建信道 var channel = connection.CreateModel(); { //创建队列 channel.QueueDeclare(queueName, durable: false, exclusive: false, autoDelete: false, arguments: null); var consumer = new EventingBasicConsumer(channel); int i = 0; consumer.Received += (model, ea) => { //消费者业务处理 var message = Encoding.UTF8.GetString(ea.Body.ToArray()); Console.WriteLine($"{i},队列{queueName}消费消息长度:{message.Length}"); i++; }; channel.BasicConsume(queueName, true, consumer); } } }

 /// 
 /// 工作队列模式 /// 
 public static void WorkerSendMsg() { string queueName = "worker_order";//队列名 //创建连接 using (var connection = RabbitMQHelper.GetConnection()) { //创建信道 using (var channel = connection.CreateModel()) { //创建队列 channel.QueueDeclare(queueName, durable: true, exclusive: false, autoDelete: false, arguments: null); 　　　　　　　　　　　　 var properties = channel.CreateBasicProperties(); 　　　　　　　　　　　　　properties.Persistent = true; //消息持久化 for ( var i=0;i<10;i++) { string message = $"Hello RabbitMQ MessageHello,{i+1}"; var body = Encoding.UTF8.GetBytes(message); //发送消息到rabbitmq channel.BasicPublish(exchange: "", routingKey: queueName, mandatory: false, basicProperties: properties, body); Console.WriteLine($"发送消息到队列:{queueName},内容:{message}"); } } } }

 public static void WorkerConsumer() { string queueName = "worker_order"; var connection = RabbitMQHelper.GetConnection(); { //创建信道 var channel = connection.CreateModel(); { //创建队列 channel.QueueDeclare(queueName, durable: false, exclusive: false, autoDelete: false, arguments: null); var consumer = new EventingBasicConsumer(channel); //prefetchCount:1来告知RabbitMQ,不要同时给一个消费者推送多于 N 个消息，也确保了消费速度和性能 channel.BasicQos(prefetchSize: 0, prefetchCount:1, global: false); int i = 1; int index = new Random().Next(10); consumer.Received += (model, ea) => { //处理业务 var message = Encoding.UTF8.GetString(ea.Body.ToArray()); Console.WriteLine($"{i},消费者:{index},队列{queueName}消费消息长度:{message.Length}"); channel.BasicAck(ea.DeliveryTag, false); //消息ack确认，告诉mq这条队列处理完，可以从mq删除了　　　　　　　　　　　　　　　Thread.Sleep(1000); i++; }; channel.BasicConsume(queueName,autoAck:false, consumer); } } }