摘要：
订单的多任务处理可以允许生产商横向扩展并支持多个服务器；生产者集群架构、消费者集群架构；在同一队列的情况下，消息被划分，并设置了平衡消费者：channel。BasicQos（0,1，false）//预取数量设置为1；设置false nuget:1。生产者使用RabbitMQMsgProducer.MessageProducer；使用Microsoft.Extensions.C
订单多任务处理
能够让生产者的横向扩展，支持多个服务器； 生产者集群架构，消费端集群架构；
同一个队列的话，消息是被瓜分掉的
设置均衡消费端：
channel.BasicQos(0, 1, false);
//预取数量设置为1个; 设置false
nuget ：
1. 生产者
using RabbitMQMsgProducer.MessageProducer; using Microsoft.Extensions.Configuration; using System; using System.IO; namespace RabbitMQMsgProducer { class Program { static void Main(string[] args) { try { { //多生产者多消费者 //订单任务处理 IConfigurationRoot config = new ConfigurationBuilder() .SetBasePath(Directory.GetCurrentDirectory()) .AddCommandLine(args)//支持命令行参数 .Build(); string strMinute = config["minute"]; //什么时候开始执行 string Num = config["num"]; //生产者编号 int minute = Convert.ToInt32(strMinute); bool flg = true; while (flg) { if (DateTime.Now.Minute == minute) { Console.WriteLine($"到{strMinute}分钟，开始写入消息。。。"); flg = false; MultiProducerMsg.Send(Num); } } } Console.ReadLine(); } catch (Exception ex) { Console.WriteLine(ex.Message); } } } }
using RabbitMQ.Client; using System; using System.Text; using System.Threading; namespace RabbitMQMsgProducer.MessageProducer { public class MultiProducerMsg { public static void Send(string Num) { ConnectionFactory connectionFactory = new ConnectionFactory(); connectionFactory.HostName = "localhost"; connectionFactory.UserName = "guest"; connectionFactory.Password = "guest"; string exchangeName = "MultiProducerMsgExchange"; string queueName = "MultiProducerMsgQueue"; using (IConnection connection = connectionFactory.CreateConnection()) { using (IModel channel = connection.CreateModel()) { // 声明 exchange 类型为 Direct，durable 持久化为true channel.ExchangeDeclare(exchange: exchangeName, type: ExchangeType.Direct, durable: true, autoDelete: false, arguments: null); // 声明 queue channel.QueueDeclare(queue: queueName, durable: true, exclusive: false, autoDelete: false, arguments: null); // 绑定 channel.QueueBind(queue: queueName, exchange: exchangeName, routingKey: string.Empty, arguments: null); Console.ForegroundColor = ConsoleColor.Red; Console.WriteLine($"the producer {Num} is ready."); while (true) { string msg = $"producer{Num} : message {DateTime.Now.ToString("yyyy-MM-dd hh:mm:ss fff")} ."; byte[] body = Encoding.UTF8.GetBytes(msg); // 发送消息 channel.BasicPublish(exchange: exchangeName, routingKey: string.Empty, basicProperties: null, body: body); Console.WriteLine($"the message : {msg} is send ."); Thread.Sleep(200); } } } } } }
2. 消费者
using RabbitMQMsgConsumer001.MessageConsumer; using System; using System.Threading.Tasks; namespace RabbitMQMsgConsumer001 { class Program { static void Main(string[] args) { try { { // 多生产多消费 // 规模处理任务清单操作 Task.Run(() => { MultiConsumerMsg.Receive001(); }); Task.Run(() => { MultiConsumerMsg.Receive002(); }); Task.Run(() => { MultiConsumerMsg.Receive003(); }); } Console.ReadLine(); } catch (Exception ex) { Console.WriteLine(ex.Message); } } } }
using RabbitMQ.Client; using RabbitMQ.Client.Events; using System; using System.Text; namespace RabbitMQMsgConsumer001.MessageConsumer { public class MultiConsumerMsg { public static void Receive001() { var factory = new ConnectionFactory(); factory.HostName = "localhost"; factory.UserName = "guest"; factory.Password = "guest"; string queueName = "MultiProducerMsgQueue"; using (var connection = factory.CreateConnection()) { using (var channel = connection.CreateModel()) { Console.ForegroundColor = ConsoleColor.Green; try { var consumer = new EventingBasicConsumer(channel); consumer.Received += (model, ea) => { var body = ea.Body; var message = Encoding.UTF8.GetString(body.ToArray()); Console.WriteLine($"the consumer 001 received: {message}"); }; channel.BasicConsume(queue: queueName, autoAck: true, consumer: consumer); Console.WriteLine(" Press [enter] to exit."); Console.ReadLine(); } catch (Exception ex) { Console.WriteLine(ex.Message); } } } } public static void Receive002() { var factory = new ConnectionFactory(); factory.HostName = "localhost"; factory.UserName = "guest"; factory.Password = "guest"; string queueName = "MultiProducerMsgQueue"; using (var connection = factory.CreateConnection()) { using (var channel = connection.CreateModel()) { Console.ForegroundColor = ConsoleColor.Green; try { var consumer = new EventingBasicConsumer(channel); consumer.Received += (model, ea) => { var body = ea.Body; var message = Encoding.UTF8.GetString(body.ToArray()); Console.WriteLine($"the consumer 002 received: {message}"); }; channel.BasicConsume(queue: queueName, autoAck: true, consumer: consumer); Console.WriteLine(" Press [enter] to exit."); Console.ReadLine(); } catch (Exception ex) { Console.WriteLine(ex.Message); } } } } public static void Receive003() { var factory = new ConnectionFactory(); factory.HostName = "localhost"; factory.UserName = "guest"; factory.Password = "guest"; string queueName = "MultiProducerMsgQueue"; using (var connection = factory.CreateConnection()) { using (var channel = connection.CreateModel()) { Console.ForegroundColor = ConsoleColor.Green; try { var consumer = new EventingBasicConsumer(channel); consumer.Received += (model, ea) => { var body = ea.Body; var message = Encoding.UTF8.GetString(body.ToArray()); Console.WriteLine($"the consumer 003 received: {message}"); }; channel.BasicConsume(queue: queueName, autoAck: true, consumer: consumer); Console.WriteLine(" Press [enter] to exit."); Console.ReadLine(); } catch (Exception ex) { Console.WriteLine(ex.Message); } } } } } }
3. 运行效果
dotnet RabbitMQMsgProducer.dll num=001 minute=22
dotnet RabbitMQMsgConsumer001.dll 