生产者
static void Main(string[] args) { var factory = new ConnectionFactory() { HostName = "192.168.1.102", Password = "guest", UserName = "guest" }; using (var connection = factory.CreateConnection()) using (var channel = connection.CreateModel()) { for (int i = 0; i < 3000; i++) { channel.QueueDeclare(queue: "test_queue", durable: true, exclusive: false, autoDelete: false, arguments: null); //设置每次发送的条数 channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false); channel.ConfirmSelect(); var body = Encoding.UTF8.GetBytes("abc" + i); var properties = channel.CreateBasicProperties(); properties.Headers = new Dictionary<string, object>(); properties.Headers.Add("name","abc") ; channel.BasicPublish("", "test_queue", properties, body); var issuccess = channel.WaitForConfirms(); } } }
消费者
static void Main(string[] args) { var factory = new ConnectionFactory() { HostName = "192.168.1.102", Password = "guest", UserName = "guest" }; using (var connection = factory.CreateConnection()) using (var channel = connection.CreateModel()) { //声明队列 channel.QueueDeclare("test_queue", true, false, false, null); //设置每次接收的条数 channel.BasicQos(0, 1, false); EventingBasicConsumer consumer = new EventingBasicConsumer(channel); consumer.Received += (sender, eventArgs) => { var body = Encoding.UTF8.GetString(eventArgs.Body.ToArray()); var headers = eventArgs.BasicProperties.Headers; headers.ToList().ForEach(x => { Console.WriteLine(x.Key + ":::" + Encoding.UTF8.GetString((byte[])x.Value)); }); Console.WriteLine(body); Thread.Sleep(1000); //手动确认,处理完成,告诉Broker可以服务端可以删除消息,分配新的消息过来 channel.BasicAck(eventArgs.DeliveryTag, false); }; //noAck设置false,告诉broker,发送消息之后,消息暂时不要删除,等消费者处理完成 channel.BasicConsume("test_queue", false, consumer); Console.WriteLine("按任意件,退出程序"); Console.ReadKey(); } }