上次我们介绍了在单机、集群下高并发场景可以选择的一些方案,传送门:高并发场景之一般解决方案
但是也发现了一些问题,比如集群下使用ConcurrentQueue或加锁都不能解决问题,后来采用Redis队列也不能完全解决问题,
因为使用Redis要自己实现分布式锁
这次我们来了解一下一个专门处理队列的组件:RabbitMQ,这个东西天生支持分布式队列。
下面我们来用RabbitMQ来实现上一篇的场景
一、新建RabbitMQ.Receive
private static ConnectionFactory factory = new ConnectionFactory { HostName = "192.168.1.109", UserName = "ljr", Password = "root", VirtualHost = "/" };
static void Main(string[] args) { using (var connection = factory.CreateConnection()) { using (var channel = connection.CreateModel()) { var consumer = new EventingBasicConsumer(); consumer.Received += (model, ea) => { var body = ea.Body; var message = Encoding.UTF8.GetString(body); Console.WriteLine(" [x] Received {0}", message); var total = DbHelper.ExecuteScalar("Select Total from ConCurrency where Id = 1", null).ToString(); ; DbHelper.ExecuteNonQuery(string.Format("Update ConCurrency Set Total = {0} where Id = 1", value.ToString()), null); }; channel.QueueDeclare(queue: "queueName", durable: false, exclusive: false, autoDelete: false, arguments: null); channel.BasicConsume(queue: "queueName", noAck: true, consumer: consumer); Console.WriteLine(" Press [enter] to exit."); Console.ReadLine(); } } }
二、新建RabbitMQ.Send
static void Main(string[] args) { ; i <= ; i++) { Task.Run(async () => { await Produce(); }); Console.WriteLine(i); } Console.ReadKey(); } public static Task Produce() { return Task.Factory.StartNew(() => { using (var connection = factory.CreateConnection()) { using (var channel = connection.CreateModel()) { var body = Encoding.UTF8.GetBytes(Guid.NewGuid().ToString()); channel.QueueDeclare(queue: "queueName", durable: false, exclusive: false, autoDelete: false, arguments: null); channel.BasicPublish(exchange: "", routingKey: "queueName", basicProperties: null, body: body); } } }); }
这里是模拟500个用户请求,正常的话最后Total就等于500
我们来说试试看,运行程序
2.1、打开接收端
2.2 运行客户端
2.3、可以看到2边几乎是实时的,再去看看数据库
好了,就是这样 、。