rabbitmq

class Program {
        private static object lockObj = new object ();

        private static Random random = new Random ();

        static void Main (string[] args) {
            Console.Title = "Consumer";
            var factory = new ConnectionFactory () { HostName = "192.168.1.7" };
            using (var connection = factory.CreateConnection ())
            using (var channel = connection.CreateModel ()) {
                channel.QueueDeclare (queue: "hello",
                    durable : true,
                    exclusive : false,
                    autoDelete : false,
                    arguments : null);

　　　　　　　　　　//设置channel给每个consumer只推一个消息, 在没有接收到consumer的ack之前, 不再给这个消费者推消息
                channel.BasicQos (prefetchSize: 0, prefetchCount: 1, global: false);

                Console.WriteLine (" [*] Waiting for messages.");

                var consumer = new EventingBasicConsumer (channel);
                consumer.Received += (sender, e) => {
                    try {
                        Console.WriteLine ($"Handle msg {e.ConsumerTag}");
                        Thread.Sleep (random.Next (100));
                        var body = e.Body;
                        var message = Encoding.UTF8.GetString (body);
                        lock (lockObj) {
                            File.AppendAllText ("./test.json", $"{message}{Environment.NewLine}");
                        }
                        Console.WriteLine ($" [{message}] handling finished!");

　　　　　　　　　　　　　　　//消息处理完后, 发送ack给channel, 以便channel给它发下一个消息来处理
                        channel.BasicAck (deliveryTag: e.DeliveryTag, multiple: false);
                    } catch (Exception ex) {
                        System.Console.WriteLine (ex.Message);
　　　　　　　　　　　　　　//消息处理失败的话, 发送nack给channel, 以但重新入队重新处理
                        channel.BasicNack (deliveryTag: e.DeliveryTag, multiple: false, requeue: true);
                    }
                };

                channel.BasicConsume (queue: "hello",
                    autoAck : false,
                    consumer : consumer);

                Console.WriteLine (" Press [enter] to exit.");
                Console.ReadLine ();
            }
        }
    }