《RabbitMQ实战指南》整理（二）客户端开发

书中以Java为例进行相关简介，这里笔者以C#为例进行相关的说明

一、连接RabbitMQ

如下通过给定的参数进行Rabbit的连接，创建之后Channel就可以用来发送或是接受消息了。需要注意的是Connection可以用来创建多个Channel实例，但是Channel不能再线程间共享，应用程序应当为每一个线程开辟一个Channel，此外多线程之间共享Channel实例是非线程安全的。

using System;
using RabbitMQ.Client;

namespace ClientDevelop
{
    class Program
    {
        static void Main(string[] args)
        {
            var factory = new ConnectionFactory
            {
                UserName = "guest",
                Password = "guest",
                HostName = "localhost"
            };
            using var conn = factory.CreateConnection();//创建连接
            using var channel = conn.CreateModel();//创建信道
        }
    }
}

通常情况下，在调用Createxxx或者newxxx方法之后，可以简单认为Connection或Channel已经成功处于开启状态，而不会使用isOpen属性来进行判断，因为该方法的返回值依赖于ShutdownCause的存在，有可能会产生竞争。

二、使用交换器和队列

交换器和队列是AMQP中high-level层面的构建模块，应用程序需要确保在使用它们的时候就已经存在了，所以在使用前需要声明它们。如下：

using System;
using RabbitMQ.Client;

namespace ClientDevelop
{
    class Program
    {
        private const string ExchangeName = "TestExchange";
        private const string RoutingKey = "TestRoute";
        
        static void Main(string[] args)
        {
            var factory = new ConnectionFactory
            {
                UserName = "guest",
                Password = "guest",
                HostName = "localhost"
            };
            using var conn = factory.CreateConnection();//创建连接
            using var channel = conn.CreateModel();//创建信道
            
            channel.ExchangeDeclare(ExchangeName,ExchangeType.Direct);//声明交换器
            var queueName = channel.QueueDeclare().QueueName;//声明队列并获取名称
            channel.QueueBind(queueName,ExchangeName,RoutingKey);//绑定交换器和队列
        }
    }
}

上述代码展示如何使用路由键将队列和交换器绑定起来，并且声明的队列具备如下特性：只对当前应用中的同一个Connection层面可用，同一个Connection的不同Channel可以共用，并且会在应用连接断开时自动删除。ExchangeDeclare和QueueDeclare方法可以根据参数的不同有不同的重载形式，可以根据自身的需要进行调整

1、ExchangeDeclare方法详解

ExchangeDeclare方法参数详细说明如下：

• exchange：交换器的名称
• type：交换器的类型，常见的有fanout、direct和topic
• durable：设置是否持久化，设置持久化可以将交换器存盘，在服务器重启时不会丢失相关的信息
• autoDelete：设置是否自动删除，删除的前提是至少一个队列或者交换器与这个交换器绑定，之后所有与这个交换器绑定的队列或者交换器都与此解绑，而不能错误理解为与此交换器连接的客户端断开时自动删除本交换器
• arguments：其他一些结构化参数

其他类似的方法如ExchangeDeclareNoWait比ExchangeDeclare多设置了一个默认值为true的noWait参数，意思为不需要服务器返回任何值。在声明完一个交换器之后，客户端紧接着使用这个交换器，必然会发生异常，因此没有特殊的缘由或场景，是不建议使用该方法的

与声明创建交换器对应的是删除交换器的方法ExchangeDelete(string exchange, bool ifUnused = false)，其中exchange为交换器的名称，ifUnused 用来设置是否在交换器没有被使用的情况下删除，true表示只有在没有被使用的情况下删除，false表示无论如何都删除

2、QueueDeclare方法详解

不带任何参数的QueueDeclare方法默认创建一个有RabbitMQ命名的排他的、自动删除的、非持久化的队列。方法参数详细说明如下：

• queue：队列的名称
• durable：设置是否持久化
• exclusive：设置是否排他。被设置为排他后，该队列仅对首次声明它的连接可见，同一连接的不同信道是可以访问的；如果一个连接已经声明了一个排他队列，其他连接是不允许建立同名的排他队列的；该队列是持久化的，一旦连接关闭或者客户端退出，该排他队列会被自动删除，因而适用于一个客户端同时发送和读取消息的场景
• autoDelete：设置为是否自动删除
• arguments：设置队列的其他一些参数

生产者和消费者都可以使用QueueDeclare来声明一个队列，但是如果消费者已经在同一个信道上订阅了另一个队列，就无法再声明队列了，必须先取消订阅，然后将信道设置为传输模式，之后才能声明队列。

和交换器一样，队列也有一个QueueDeclareNoWait方法，同样也需要注意声明完紧接着使用会发生异常的情况

和交换器一样，队列也有对应的删除方法QueueDelete(string queue, bool ifUnused = false, bool ifEmpty = false)，ifEmpty设置为true表示在队列为空的情况下才能够删除

3、QueueBind方法详解

• queue：队列名称
• exchange：交换器的名称
• routingKey：用来绑定队列和交换器的路由键
• arguments：定义绑定的一些参数

4、ExchangeBind方法详解

不仅可以将交换器与队列绑定，也可以将交换器与交换器绑定，方法参数与ExchangeDeclare方法类似。绑定之后Source交换器会将消息转发到Destination交换器，某种程度上来说Destination交换器可以看作一个队列，示例如下：

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

namespace ClientDevelop
{
    class Program
    {
        private const string QueueName = "TestQueue";
        private const string RoutingKey = "TestRoute";

        static void Main(string[] args)
        {
            var factory = new ConnectionFactory
            {
                UserName = "guest",
                Password = "guest",
                HostName = "localhost"
            };
            using var conn = factory.CreateConnection();//创建连接
            using var channel = conn.CreateModel();//创建信道
            
            channel.ExchangeDeclare("source", ExchangeType.Direct, false, true, null);//声明交换器1
            channel.ExchangeDeclare("destination", ExchangeType.Fanout, false, true, null);//声明交换器2
            channel.ExchangeBind("destination", "source", RoutingKey);//绑定两个交换器
            
            channel.QueueDeclare(QueueName, false, false, true, null);//声明队列
            channel.QueueBind(QueueName, "destination", RoutingKey);//绑定队列
            
            channel.BasicPublish("source", RoutingKey, null, Encoding.UTF8.GetBytes("exToExDemo"));//发布消息
        }
    }
}

5、何时创建

RabbitMQ的消息存储在队列中，交换器的使用并不耗费服务器的性能，而队列会，因此衡量RabbitMQ当前的QPS只需要看队列即可。按照官方建议，生产者和消费者都应该尝试创建队列，这是一个很好的建议但并不适用于所有的情况。在一些已经充分预估了队列的使用情况下，完全可以先创建好而不是在业务代码中声明，同时这样做的好处是避免匹配异常的情况

三、发送消息

如果要发送一条消息，可以使用Channel的BasicPublish方法，为了更好的控制发送，可以使用Mandatory参数，或者使用IModel类的CreateBasicProperties方法发送一些特定属性的信息，常用的参数详细说明如下：

• exchange：交换器的名称
• routingKey：路由键，交换器根据路由键将消息存储到对应的队列中
• basicProperties：消息的基本属性集，包含许多属性成员
• body：消息体，真正需要发送的消息

四、消费消息

RabbitMQ的消费模式有两种，推模式和拉模式，推模式采用Basic.Consume进行消费，拉模式则调用Basic.Get进行消费

1、推模式

在推模式中，可以使用持续订阅模式来消费消息，不同的订阅采用不同的消费者标签来区分彼此，在同一个信道中的消费者也需要通过唯一的消费者标签以作区分

服务端：

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

namespace Producer
{
    class Program
    {
        private const string ExchangeName = "TestExchange";
        private const string QueueName = "TestQueue";
        private const string RoutingKey = "TestRoute";

        static void Main(string[] args)
        {
            var factory = new ConnectionFactory
            {
                HostName = "localhost",
                UserName = "guest",
                Password = "guest"
            };

            using var conn = factory.CreateConnection();//创建连接
            using var channel = conn.CreateModel();//创建信道

            channel.ExchangeDeclare(ExchangeName, ExchangeType.Direct);//声明交换器
            channel.QueueDeclare(QueueName, false, false, false, null);//声明队列
            channel.QueueBind(QueueName, ExchangeName, RoutingKey,null);//绑定交换器和队列

            //发布消息
            const string message = "Hello World";
            channel.BasicPublish(ExchangeName, RoutingKey, null, Encoding.UTF8.GetBytes(message));
            Console.WriteLine(" [x] Sent {0}", message);

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

客户端：

using System;
using System.Text;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

namespace Consumer
{
    class Program
    {
        private const string ExchangeName = "TestExchange";
        private const string QueueName = "TestQueue";
        private const string RoutingKey = "TestRoute";
        private const string ConsumerTag = "TestTag";

        static void Main(string[] args)
        {
            var factory = new ConnectionFactory
            {
                HostName = "localhost",
                UserName = "guest",
                Password = "guest"
            };

            using var connection = factory.CreateConnection();
            using var channel = connection.CreateModel();

            channel.ExchangeDeclare(ExchangeName, ExchangeType.Direct);//声明交换器
            channel.QueueDeclare(QueueName, false, false, false, null);//声明队列
            channel.QueueBind(QueueName, ExchangeName, RoutingKey, null);//绑定交换器和队列

            Console.WriteLine("Waiting for message...");

            //推模式处理数据
            channel.BasicQos(0, 1, false); //未收到消费端确认时不再分发消息
            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (model, ea) =>
            {
                var body = ea.Body.ToArray();
                var message = Encoding.UTF8.GetString(body);
                Console.WriteLine(" [x] {0}", message);
                channel.BasicAck(ea.DeliveryTag, false);
            };
            channel.BasicConsume(QueueName, false, ConsumerTag, consumer);

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

在BasicConsume方法中我们显式地设置autoAck为false，然后在接受到消息后进行显式ack操作，这样做可以防止消息不必要的丢失。BasicConsume方法有多个重载，常用的参数说明如下：

• queue：队列的名称
• autoAck：设置是否自动确认，建议设置成false；
• consumerTag：消费者标签，用来区分多个消费者；
• noLocal：设置成true则表示不能将同一个Connection中生产者发送的消息传递给这个Connection中的消费者；
• exclusive：设置是否排他；
• arguments：设置消费者的其他参数
• consumer：设置消费者的回调函数，用来处理RabbitMQ推送过来的消息，比如DefaultConsumer

和生产者一样，消费者客户端同样需要考虑线程安全的问题，消费者客户端的callback会被分配到Channel不同的线程上，这意味者消费者客户端可以安全地调用这些阻塞的方法。最常用的做法是一个Channel对应一个消费者，若存在多个那么其他消费者的callback会被阻塞。

2、拉模式

拉模式通过channel.basicGet方法可以单条地获取消息，其返回值是GetRespone。如果autoAck为false，那么同样需要调用channel.BasicAck来确认消息被接受

using System;
using System.Text;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

namespace Consumer
{
    class Program
    {
        private const string ExchangeName = "TestExchange";
        private const string QueueName = "TestQueue";
        private const string RoutingKey = "TestRoute";
        private const string ConsumerTag = "TestTag";

        static void Main(string[] args)
        {
            var factory = new ConnectionFactory
            {
                HostName = "localhost",
                UserName = "guest",
                Password = "guest"
            };

            using var connection = factory.CreateConnection();
            using var channel = connection.CreateModel();

            channel.ExchangeDeclare(ExchangeName, ExchangeType.Direct);//声明交换器
            channel.QueueDeclare(QueueName, false, false, false, null);//声明队列
            channel.QueueBind(QueueName, ExchangeName, RoutingKey, null);//绑定交换器和队列

            Console.WriteLine("Waiting for message...");

            //拉模式处理数据
            var res = channel.BasicGet(QueueName, false);
            Console.WriteLine(Encoding.Default.GetString(res.Body.ToArray()));
            channel.BasicAck(res.DeliveryTag, false);

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

注：Basic.Consume将信道设置为投递模式，直到取消队列的订阅为止，在投递期间，RabbitMq会不断地推送消息给消费者。如果只想从队列获得单条消息，可以使用Basic.Get进行消费，但不能将其放在循环中替代Basic.Consume，这样做会严重影响性能。要实现高吞吐量，消费者理应使用Basic.Consume方法

五、消费端的确认与拒绝

为保证消息从队列可靠地到达消费者，RabbitMQ提供了消息确认机制。消费者在订阅队列时，可以指定autoAck参数，当其为false时，RabbitMQ会等待消费者显式地回复信号后才从内存或磁盘中移除消息（实际上是先打上删除标记再删除）；此外对于服务端而言，队列中的消息会分为两种，一种是等待传递给消费者的消息，一种是已经传递给消费者的消息，如果RabbitMQ一直没有收到消费者的确认信号，并且消息的消费者已经断开连接，RabbitMQ会安排消息重新进入队列进行投递。RabbitMQ判断是否需要重新已投递的唯一依据是某消息的对应的消费者是否已经断开。可以参照Web管理页面中的Ready和Uncaked字段进行查看

如果像明确拒绝当前的消息而不是确认，可以使用BasicReject(ulong deliveryTag, bool requeue)命令，其中deliverTag是消息的编号，requeue为true则RabbitMQ会重新将这条消息存入队列，以便发送给下一个订阅的消费者，如果为false该消息会被移除。BasicReject命令一次只能拒绝一条消息，如果想要批量拒绝，可以使用BasicNack(ulong deliveryTag, bool multiple, bool requeue)方法，multiple参数为false时表示拒绝编号为deliveryTag的这一条消息，为true时表示拒绝deliveryTag编号之前所有未被当前消费者确认的消息。

BasicRecover(bool requeue)方法可以用来请求RabbitMQ重新发送给未被确认的消息，requeue为true时未被确认的消息会被重新加入到队列中，如果为false会被分配给与之前相同的消费者

六、关闭连接

Connection关闭时，Channel也会自动关闭。AMQP协议中的Connection和Channel采用相同的方式来管理网络失败、内部错误和显式地关闭连接，两者的生命周期如下：

• Open：开启状态，代表当前对象可用；
• Closing：正在关闭的状态，当前对象被显式的调用关闭方法时会产生关闭请求对内部对象进行相应的操做，并等待这些关闭操作的完成；
• Closed：已经关闭的状态，当前对象已经接受到所有内部对象已完成关闭动作的通知，并且其自身也已关闭