C#之RabbitMQ

using System;
using System.Text;
using RabbitMQ.Client;123

class Send
{
    public static void Main()
    {
        var factory = new ConnectionFactory() { HostName = "localhost" };//创建代理服务器实例。注意：HostName为Rabbit Server所在服务器的ip或域名，如果服务装在本机，则为localhost，默认端口5672
        using (var connection = factory.CreateConnection())//创建socket连接
        {
            using (var channel = connection.CreateModel())//channel中包含几乎所有的api来供我们操作queue
            {
                ...
            }
        }
    }
}

using System;
using RabbitMQ.Client;
using System.Text;
​
class Send
{
    public static void Main()
    {
        var factory = new ConnectionFactory() { HostName = "localhost" };
        using(var connection = factory.CreateConnection())
        using(var channel = connection.CreateModel())
        {
            //声明queue
            channel.QueueDeclare(queue: "hello",//队列名
                                 durable: false,//是否持久化
                                 exclusive: false,//true:排他性，该队列仅对首次申明它的连接可见，并在连接断开时自动删除
                                 autoDelete: false,//true:如果该队列没有任何订阅的消费者的话，该队列会被自动删除
                                 arguments: null);//如果安装了队列优先级插件则可以设置优先级
​
            string message = "Hello World!";//待发送的消息
            var body = Encoding.UTF8.GetBytes(message);
​
            channel.BasicPublish(exchange: "",//exchange名称
                                 routingKey: "hello",//如果存在exchange,则消息被发送到名称为hello的queue的客户端
                                 basicProperties: null,
                                 body: body);//消息体
            Console.WriteLine(" [x] Sent {0}", message);
        }
​
        Console.WriteLine(" Press [enter] to exit.");
        Console.ReadLine();
    }
}

using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Text;
​
class Receive
{
    public static void Main()
    {
        var factory = new ConnectionFactory() { HostName = "localhost" };
        using(var connection = factory.CreateConnection())
        using(var channel = connection.CreateModel())
        {
            channel.QueueDeclare(queue: "hello",//指定发送消息的queue，和生产者的queue匹配
                                 durable: false,
                                 exclusive: false,
                                 autoDelete: false,
                                 arguments: null);
​
            var consumer = new EventingBasicConsumer(channel);
            //注册接收事件，一旦创建连接就去拉取消息
            consumer.Received += (model, ea) =>
            {
                var body = ea.Body;
                var message = Encoding.UTF8.GetString(body);
                Console.WriteLine(" [x] Received {0}", message);
            };
            channel.BasicConsume(queue: "hello",
                                 noAck: true,//和tcp协议的ack一样，为false则服务端必须在收到客户端的回执（ack）后才能删除本条消息
                                 consumer: consumer);
​
            Console.WriteLine(" Press [enter] to exit.");
            Console.ReadLine();
        }
    }
}

using System;
using System.Collections.Generic;
using System.Text;
using RabbitMQ.Client;
​
namespace NewTask
{
    class NewTask
    {
        public static void Main(string[] args)
        {
            var factory = new ConnectionFactory() { HostName = "localhost" };
            using (var connection = factory.CreateConnection())
            using (var channel = connection.CreateModel())
            {
                channel.QueueDeclare(queue: "task_queue",
                                     durable: true,
                                     exclusive: false,
                                     autoDelete: false,
                                     arguments: null);
                //At this point we're sure that the task_queue queue won't be lost even if RabbitMQ restarts. Now we need to mark our messages as persistent - by setting IBasicProperties.SetPersistent to true.
                var properties = channel.CreateBasicProperties();
                //properties.SetPersistent(true);//此方法已经过时:“这种设置方法已经被摒弃，现在使用持久化属性取代它”
                properties.Persistent = true;//取代上面的写法
​
                ...
        }
    }
}

using System;
using System.Text;
using System.Threading;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
​
namespace Worker
{
    class Worker
    {
        public static void Main()
        {
            var factory = new ConnectionFactory() { HostName = "localhost" };
            using (var connection = factory.CreateConnection())
            using (var channel = connection.CreateModel())
            {
                channel.QueueDeclare(queue: "task_queue",
                                     durable: true,
                                     exclusive: false,
                                     autoDelete: false,
                                     arguments: null);
​
                channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);//prefetchCount:This tells RabbitMQ not to give more than one message to a worker at a time. Or, in other words, don't dispatch a new message to a worker until it has processed and acknowledged the previous one. Instead, it will dispatch it to the next worker that is not still busy（Fair dispatch（公平分发）：它会选择空闲的worker去处理消息，适用于对性能敏感的场景；Round-robin dispatching（轮循分发）：不解释）
​
                Console.WriteLine(" [*] Waiting for messages.");
​
                var consumer = new EventingBasicConsumer(channel);
                channel.BasicConsume(queue: "task_queue",
                                     noAck: false,
                                     consumer: consumer);
                consumer.Received += (model, ea) =>
                {
                    var body = ea.Body;
                    var message = Encoding.UTF8.GetString(body);
                    Console.WriteLine(" [x] Received {0}", message);
​
                    int dots = message.Split('.').Length - 1;
                    Thread.Sleep(dots * 1000);
​
                    Console.WriteLine(" [x] Done");
​
                    channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);//用以处理完毕后发送ack通知服务端
                };
​
                Console.WriteLine(" Press [enter] to exit.");
                Console.ReadLine();
            }
        }
    }
}