RabbitMQ基本实践

声明：本文是《RabbitMQ实战指南》（朱忠华 著）学习笔记，仅供本人学习研究之用，如若喜欢请购买正版书籍。如有侵权，请联系删除。

---

连接 RabbitMQ

下面的代码用来在给定的参数（IP地址、端口号、用户名、密码等）下连接 RabbitMQ：

 

也可以使用 URI 的方式来实现：

 

获取到Connection对象后，可以使用它来创建一个Channel：

 

获取到Channel之后，可以用它来发送或者接收消息了。

Connection 可以用来创建多个 Channel 实例，但是 Channel 实例不能在线程问共享，应用程序应该为每一个线程开辟一个 Channel，也就是 Channel 实例是非线程安全的。

Channel 或者 Connection 中有个 isOpen 方法可以用来检测其是否己处于开启状态，但并不推荐在生产环境的代码上使用 isOpen 方法，这个方法的返回值依赖于 shutdownCause 的存在，有可能会产生竞争。通常情况下，在调用 createXXX 或者 newXXX 方法之后，可以简单地认为 Connection 或者 Channel 已经处于开启状态，而并不会在代码中使用 isOpen 这个检测方法。如果在使用 Channel 的时候其己经处于关闭状态，那么程序会抛出一个 com.rabbitmq.client.ShutdownSignalException，我们只需捕获这个异常即可。当然同时也要试着捕获 IOExceptio 口或者 SocketException，以防 Connection 意外关闭。

使用交换器和队列

使用交换器和队列前要先声明（declare）它们，下面的代码演示了如何声明一个交换器和队列：

 

上面的代码创建了一个持久化的、非自动删除的、绑定类型为 direct 的交换器，同时也创建了一个非持久化的、排他的、自动删除的队列（队列的名称由 RabbitMQ 自动生成），展示了如何使用路由键将队列和交换器绑定起来。上面声明的队列具备如下特性：只对当前应用中同一个 Connection 可用，同一个 Connection 的不同 Channel 可共用，并且也会在应用连接断开时自动删除。

如果要在应用中共享一个队列，可以做如下声明：

 

这里的队列被声明为持久化的、非排他的、非自动删除的，而且有一个确定的己知的名称（由客户端分配而非 RabbitMQ 自动生成）。

生产者和消费者都可以声明一个交换器或者队列。如果尝试声明一个已经存在的交换器或者队列，只要声明的参数完全匹配现存的交换器或者队列，RabbitMQ 就什么都不做，并成功返回。如果声明的参数不匹配则会抛出异常。

声明和删除交换器

exchangeDeclare 用于声明交换器，有多个重载方法，返回值类型都是DeclareOk，用来标识成功声明了一个交换器。如下：

 

方法参数如下：

• exchangeName：交换器的名称。

• type：交换器的类型，常见的如 fanout、direct、topic等。

• durable：是否持久化。true 表示持久化，false 表示非持久化。

• autoDelete：是否自动删除。为 true 表示自动删除。自动删除的前提是至少有一个队列或者交换器与这个交换器绑定，之后所有与这个交换器绑定的队列或者交换器都与此解绑。注意不能错误地把这个参数理解为：当与此交换器连接的客户端都断开时，RabbitMQ 会自动删除本交换器 。

• internal：是否是内置的。如果设置为 true ，则表示是内置的交换器，客户端程序无法直接发送消息到这个交换器中，只能通过交换器路由到交换器这种方式。

• arguments：其他一些结构化参数，比如 alternate-exchange。

除了 exchangeDeclare 之外，还有 exchangeDeclareNoWait 方法：

 

它比 exchangeDeclare 多设置了一个 nowait 参数 ，这个 nowait 参数指的是 AMQP 中 Exchange.Declare 命令的参数，意思是不需要服务器返回。这个方法的返回值是 void ，而 exchangeDeclare 方法的返回值是 Exchange.DeclareOk，意思是在客户端声明了一个交换器之后，需要等待服务器的返回（服务器会返回 Exchange.Declare-Ok 这个 AMQP 命令)。

如果使用 exchangeDeclareNoWait 方法声明完一个交换器之后（实际服务器还并未完成交换器的创建），客户端马上使用这个交换器，必然会发生异常。如果没有特殊的缘由和应用场景，并不建议使用这个方法。

除了 exchangeDeclare 和 exchangeDeclareNoWait，还有 exchangeDeclarePassive 方法：

 

这个方法主要用来检测相应的交换器是否存在。如果存在则正常返回，如果不存在则抛出异常：404 channel exception，同时 Channel 也会被关闭。

交换器可以删除，删除交换器的方法如下：

 

其中 exchangeName 表示交换器的名称，ifUnused 用来设置是否在交换器没有被使用的情况下删除。如果设置为 true，则只有在此交换器没有被使用的情况下才会被删除，如果设置 false，则无论如何这个交换器都要被删除。

声明和删除队列

queueDeclare 用于声明队列，其重载方法如下：

 

不带任何参数的 queueDeclare 方法默认创建一个由 RabbitMQ 命名的（即匿名队列）、排他的、自动删除的、非持久化的队列。

带参数的方法，参数说明如下：

• queue：队列的名称。

• durable：是否持久化。true 表示持久化，false 表示非持久化。

• exclusive：是否排他。为 true 则设置队列为排他的。如果一个队列被声明为排他队列，该队列仅对首次声明它的连接可见，并在连接断开时自动删除。注意三点：排他队列是基于连接（Connection）可见的，同一个连接的不同信道（Channel）是可以同时访问同一连接创建的排他队列； 首次是指如果一个连接己经声明了一个排他队列，其他连接是不允许建立同名的排他队列的，这个与普通队列不同；即使该队列是持久化的，一旦连接关闭或者客户端退出，该排他队列都会被自动删除，这种队列适用于一个客户端同时发送和读取消息的应用场景。

• autoDelete：是否自动删除。为 true 表示自动删除。自动删除的前提是：至少有一个消费者连接到这个队列，之后所有与这个队列连接的消费者都断开时，才会自动删除。并不是当连接到此队列的所有客户端断开时，这个队列自动删除，因为生产者客户端创建这个队列，或者没有消费者客户端与这个队列连接时，都不会自动删除这个队列。

• arguments：设置队列的其他一些参数，如 x-message-ttl、x-expires 等。

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

除了 queueDeclare，还有 queueDeclareNoWait 方法：

 

方法的返回值也是 void ，表示不需要服务端的任何返回。同样也需要注意，在调用完 queueDeclareNoWait 方法之后，紧接着使用声明的队列时有可能会发生异常情况。

还有一个 queueDeclarePassive 方法：

 

这个方法用来检测相应的队列是否存在。如果存在则正常返回，如果不存在则抛出异常：404 channel exception，同时 Channel 也会被关闭。

删除队列的相应方法如下：

 

queue 表示队列的名称，ifUnused 用来设置是否在队列没有被使用的情况下删除。 ifEmpty 设置为true 表示在队列为空（队列里面没有任何消息堆积）的情况下才能够删除。

与队列相关的还有一个方法 queuePurge，区别于 queueDelete，这个方法用来清空队列中的内容，而不删除队列本身：

 

queue 是队列的名称。

将队列和交换器绑定或解绑

queueBind 用于将队列和交换器绑定：

 

参数如下：

• queue：队列名称。

• exchange：交换器名称。

• routingKey：路由键。

• arguments：定义绑定的一些参数。

可以将已经被绑定的队列和交换器进行解绑，方法如下：

 

参数同 queueBind。

交换器与交换器绑定或解绑

不仅可以将交换器与队列绑定，也可以将交换器与交换器绑定，相应的方法如下：

 

• destination：目的地交换器。

• source：源交换器。

• routingKey：路由键。

• auguments：一些参数。

绑定之后，消息从 source 交换器转发到 destination 交换器。示例代码如下：

 

生产者发送消息至交换器 source 中，交换器 source 根据路由键找到与其匹配的另一个交换器 destination，并把消息转发到 destination 中，进而存储在与 destination 绑定的队列 queue 中。

发送消息

可以使用 Channel 类的 basicPublish 方法发送消息，basicPublish 方法如下：

 

• exchange：交换器的名称，指明消息需要发送到哪个交换器中。如果设置为空字符串，则消息会被发送到 RabbitMQ 默认的交换器中。

• routingkey：路由键，交换器根据路由键将消息存储到相应的队列之中。

• props：消息的基本属性集，包含 14 个属性成员，分别有 contentType、contentEncoding、headers ( Map<String，Object>) 、 deliveryMode、priority、correlationld、replyTo、expiration、messageld、timestamp、type、userld、appld、clusterId。

• body：消息体（payload），真正需要发送的消息。

• mandatory 和 immediate 后面介绍。

比如发送一条内容为"Hello World!"的消息：

 

为了更好地控制发送，可以使用 mandatory 这个参数 ， 或者可以发送一些特定属性的信息：

 

上面这行代码发送了一条消息，这条消息的投递模式 ( delivery mode ) 设置为 2 ，即消息会被持久化在服务器中。同时这条消息的优先级（priority）设置为 0，content-type为 "text/plain"。可以自己设定消息的属性：

 

也可以发送一条带有 headers 的消息：

 

还可以发送一条带有过期时间 (expiration ) 的消息：

 

消费消息

RabbitMQ 的消费模式分两种：推（Push）模式和拉（Pull）模式。推模式采用 Basic.Consume 进行消费，而拉模式则是调用 Basic.Get 进行消费。

推模式

在推模式中，可以通过持续订阅的方式来消费消息，使用到的相关类有：

 

接收消息一般通过实现 Consumer 接口或者继承 DefaultConsumer 类来实现。当调用与 Consumer 相关的 API 方法时，不同的订阅采用不同的消费者标签（consumerTag）来区分彼此，在同一个 Channel 中的消费者也需要通过唯一的消费者标签以作区分，消费者代码如下：

 

上面代码中显式地设置 autoAck 为 false，然后在接收到消息之后进行显式 ack 操作 (channel.basicAck )，对于消费者来说这个设置是非常必要的，可以防止消息不必要地丢失。

basicConsume方法有以下几种形式：

 

• queue：队列名称。

• autoAck：是否自动确认。

• consumerTag：消费者标签，用来区分多个消费者。

• noLocal：设置为 true 则表示不能将同一个 Connection 中生产者发送的消息传送给这个 Connection 中的消费者。

• exclusive：是否排他。

• arguments：设置其他参数。

• callback：设置消费者的回调函数，用来处理 RabbitMQ 推送过来的消息，比如 DefaultConsumer，使用时需要客户端重写其中的方法。

对于消费者客户端来说重写 handleDelivery 方法是十分方便的。更复杂的消费者客户端会重写更多的方法，具体如下：

 

handleShutdownSignal 方法，当 Channel 或者 Connection 关闭的时候会调用。handleConsumeOk 方法会在其他方法之前调用，返回消费者标签。

重写 handleCancelOk 和 handleCancel 方法，这样消费端可以在显式地或者隐式地取消订阅的时候调用。也可以通过 channel.basicCancel 方法来显式地取消一个消费者的订阅：

 

上面这行代码会首先触发 handleConsumerOk 方法，之后触发 handleDelivery 方法，最后才触发 handleCancelOk 方法。

和生产者一样，消费者客户端同样需要考虑线程安全的问题。消费者客户端的这些 callback 会被分配到与 Channel 不同的线程池上，这意味着消费者客户端可以安全地调用这些阻塞方法，比如 channel.queueDeclare、 channel.basicCancel 等。

每个 Channel 都拥有自己独立的线程。最常用的做法是一个 Channel 对应一个消费者，也就是意味着消费者彼此之间没有任何关联。当然也可以在一个 Channel 中维持多个消费者，但是要注意一个问题，如果 Channel 中的一个消费者一直在运行，那么其他消费者的 callback会被"耽搁"。

拉模式

通过 channel.basicGet 方法可以单条地获取消息，其返回值是 GetResponse。Channel 类的 basicGet 方法没有其他重载方法：

 

参数中，queue 代表队列的名称，如果设置 autoAck 为 false ， 那么同样需要调用 channel.basicAck 来确认消息己被成功接收。

示例代码：

 

消费端的确认与拒绝

为了保证消息从队列可靠地达到消费者，RabbitMQ 提供了消息确认机制（message acknowledgement）。消费者在订阅队列时，可以指定 autoAck 参数，当 autoAck 等于 false 时，RabbitMQ 会等待消费者显式地回复确认信号后才从内存（或者磁盘）中移去消息。当 autoAck 等于 true 时，RabbitMQ 会自动把发送出去的消息置为确认，然后从内存（或者磁盘）中删除，而不管消费者是否真正地消费到了这些消息 。

采用消息确认机制后，只要设置 autoAck 参数为 false，消费者就有足够的时间处理消息，不用担心处理消息过程中消费者进程挂掉后消息丢失的问题，因为 RabbitMQ 会一直持有消息直到消费者显式调用 Basic.Ack 命令为止。

当 autoAck 参数置为 false，对于 RabbitMQ 服务端而言，队列中的消息分成了两个部分：一部分是等待投递给消费者的消息，一部分是己经投递给消费者，但是还没有收到消费者确认信号的消息。如果 RabbitMQ 一直没有收到消费者的确认信号，并且消费此消息的消费者己经断开连接，则 RabbitMQ 会安排该消息重新进入队列，等待投递给下一个消费者，当然也有可能还是原来的那个消费者。

RabbitMQ 不会为未确认的消息设置过期时间，它判断此消息是否需要重新投递给消费者的唯一依据是消费该消息的消费者连接是否己经断开，这么设计的原因是 RabbitMQ 允许消费者消费一条消息的时间可以很久很久。

可以使用如下命令查看当前队列中的 "Ready" 状态和 "Unacknowledged" 状态的消息数，分别对应上文中的等待投递给消费者的消息数和己经投递给消费者但是未收到确认信号的消息数：

 

也可以在 RabbtiMQ 的 Web 管理平台查看这些信息。

消费者接收到消息后，如果想明确拒绝当前的消息而不是确认，那么应该怎么做呢？RabbitMQ 在 2 .0.0 版本开始引入了 Basic.Reject 这个命令，消费者客户端可以调用与其对应的 channel.basicReject 方法来告诉 RabbitMQ 拒绝这个消息 。

Channel 中 basicReject 方法定义如下：

 

其中 deliveryTag 可以看作消息的编号，它是一个 64 位的长整型值，最大值是 9223372036854775807。如果 requeue 参数设置为 true ，则 RabbitMQ 会重新将这条消息存入队列，以便可以发送给下一个订阅的消费者；如果 requeue 参数设置为 false，则 RabbitMQ 会把消息从队列中移除，而不会把它发送给新的消费者。

Basic.Reject 命令一次只能拒绝一条消息，如果想要批量拒绝消息，则可以使用 Basic.Nack 这个命令。消费者客户端可以调用 channel.basicNack 方法来实现，方法定义如下：

 

其中 deliveryTag 和 requeue 的含义可以参考 basicReject 方法。multiple 参数设置为 false 则表示拒绝编号为 deliveryTag 的这 一条消息，这时候 basicNack 和 basicReject 方法一样；multiple 参数设置为 true 则表示拒绝 deliveryTag 编号之前所有未被当前消费者确认的消息。

关闭连接

在应用程序使用完之后，需要关闭连接，释放资源：

 

from:https://blog.csdn.net/qmqm011/article/details/83824378