RabbitMQ 简述
RabbitMQ是一个消息代理:它接受并转发消息。 您可以将其视为邮局:当您将要把寄发的邮件投递到邮箱中时,您可以确信Postman 先生最终会将邮件发送给收件人。 在这个比喻中,RabbitMQ是一个邮箱,邮局和邮递员,用来接受,存储和转发二进制数据块的消息。
队列就像是在RabbitMQ中扮演邮箱的角色。 虽然消息经过RabbitMQ和应用程序,但它们只能存储在队列中。 队列只受主机的内存和磁盘限制的限制,它本质上是一个大的消息缓冲区。 许多生产者可以发送到一个队列的消息,许多消费者可以尝试从一个队列接收数据。
producer即为生产者,用来产生消息发送给队列。consumer是消费者,需要去读队列内的消息。producer,consumer和broker(rabbitMQ server)不必驻留在同一个主机上;确实在大多数应用程序中它们是这样分布的。
简单队列
简单队列是最简单的一种模式,由生产者、队列、消费者组成。生产者将消息发送给队列,消费者从队列中读取消息完成消费。
在下图中,“P”是我们的生产者,“C”是我们的消费者。 中间的框是队列 - RabbitMQ代表消费者的消息缓冲区。
java 方式
生产者
public class MyProducer {
private static final String QUEUE_NAME = "ITEM_QUEUE";
public static void main(String[] args) throws Exception {
// 1. 创建一个 ConnectionFactory 并进行设置
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setVirtualHost("/");
factory.setUsername("guest");
factory.setPassword("guest");
// 2. 通过连接工厂来创建连接
Connection connection = factory.newConnection();
// 3. 通过 Connection 来创建 Channel
Channel channel = connection.createChannel();
/* Message 消息 Properties
* deliverMode 设置为 2 的时候代表持久化消息
* expiration 意思是设置消息的有效期,超过10秒没有被消费者接收后会被自动删除
* headers 自定义的一些属性
*/
Map<String, Object> headers = new HashMap<>();
headers.put("myHead1", "111");
headers.put("myHead2", "222");
AMQP.BasicProperties properties =
new AMQP.BasicProperties()
.builder()
.deliveryMode(2)
.contentEncoding("UTF-8")
.expiration("100000")
.headers(headers)
.build();
// 实际场景中,消息多为json格式的对象 Message:消息的 Body
String msg = "hello";
// 4. 发送三条数据
for (int i = 1; i <= 3; i++) {
channel.basicPublish("", QUEUE_NAME, properties, msg.getBytes());
System.out.println("Send message" + i + " : " + msg);
}
// 5. 关闭连接
channel.close();
connection.close();
}
}
消费者
public class MyConsumer {
private static final String QUEUE_NAME = "ITEM_QUEUE";
public static void main(String[] args) throws Exception {
// 1. 创建一个 ConnectionFactory 并进行设置
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setVirtualHost("/");
factory.setUsername("guest");
factory.setPassword("guest");
// 2. 通过连接工厂来创建连接
Connection connection = factory.newConnection();
// 3. 通过 Connection 来创建 Channel
Channel channel = connection.createChannel();
// 4. 声明一个队列
/* durable 持久化
* exclusive 排他队列
* 如果你想创建一个只有自己可见的队列,即不允许其它用户访问,RabbitMQ允许你将一个Queue声明成为排他性的(Exclusive Queue)。
* 该队列的特点是:
** 只对首次声明它的连接(Connection)可见
** 会在其连接断开的时候自动删除。
** 对于第一点,首先是强调首次声明,因为另外一个连接无法声明一个同样的排他性队列;其次是只区别连接(Connection)而不是通道(Channel),从同一个连接创建的不同的通道可以同时访问某一个排他性的队列。这里说的连接是指一个AMQPConnection,以RabbitMQ的Java客户端为例:
** 如果试图在一个不同的连接中重新声明或访问(如publish,consume)该排他性队列,会得到资源被锁定的错误:
** `ESOURCE_LOCKED - cannot obtain exclusive access to locked queue 'UserLogin2'`
** 对于第二点,RabbitMQ会自动删除这个队列,而不管这个队列是否被声明成持久性的(Durable =true)。 也就是说即使客户端程序将一个排他性的队列声明成了Durable的,只要调用了连接的Close方法或者客户端程序退出了,RabbitMQ都会删除这个队列。注意这里是连接断开的时候,而不是通道断开。这个其实前一点保持一致,只区别连接而非通道。
* */
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
/*
true:表示自动确认,只要消息从队列中获取,无论消费者获取到消息后是否成功消费,都会认为消息已经成功消费
false:表示手动确认,消费者获取消息后,服务器会将该消息标记为不可用状态,等待消费者的反馈,如果消费者一
直没有反馈,那么该消息将一直处于不可用状态,并且服务器会认为该消费者已经挂掉,不会再给其发送消息,
直到该消费者反馈。
*/
// 5. 创建消费者并接收消息
Consumer consumer =
new DefaultConsumer(channel) {
@Override
public void handleDelivery(
String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
throws IOException {
String message = new String(body, "UTF-8");
Map<String, Object> headers = properties.getHeaders();
System.out.println("head: " + headers.get("myHead1"));
System.out.println(" [x] Received '" + message + "'");
System.out.println("expiration : " + properties.getExpiration());
}
};
// 6. 设置 Channel 消费者绑定队列
channel.basicConsume(QUEUE_NAME, true, consumer);
}
}
运行结果
## MyProducer
Send message1 : hello
Send message2 : hello
Send message3 : hello
## MyConsumer
[*] Waiting for messages. To exit press CTRL+C
head: 111
[x] Received 'hello'
expiration : 100000
head: 111
[x] Received 'hello'
expiration : 100000
head: 111
[x] Received 'hello'
expiration : 100000
当我们启动生产者之后查看RabbitMQ管理后台可以看到有一条消息正在等待被消费(需要先手动创建队列或者启动一下消费者)。
当我们启动消费者之后再次查看,可以看到积压的一条消息已经被消费。
总结
- 声明交换机
/**
* Declare an exchange, via an interface that allows the complete set of
* arguments.
* @see com.rabbitmq.client.AMQP.Exchange.Declare
* @see com.rabbitmq.client.AMQP.Exchange.DeclareOk
* @param exchange the name of the exchange
* @param type the exchange type
* @param durable true if we are declaring a durable exchange (the exchange will survive a server restart)
* @param autoDelete true if the server should delete the exchange when it is no longer in use
* @param internal true if the exchange is internal, i.e. can't be directly
* published to by a client.
* @param arguments other properties (construction arguments) for the exchange
* @return a declaration-confirm method to indicate the exchange was successfully declared
* @throws java.io.IOException if an error is encountered
*/
Exchange.DeclareOk exchangeDeclare(String exchange,
String type,boolean durable,
boolean autoDelete,boolean internal,
Map<String, Object> arguments) throws IOException;
-
Name: 交换机名称
-
Type: 交换机类型direct、topic、 fanout、 headers
-
Durability: 是否需要持久化,true为持久化
-
Auto Delete: 当最后一个绑定到Exchange. 上的队列删除后,自动删除该Exchange
-
Internal: 当前Exchange是否用于RabbitMQ内部使用,默认为False
-
Arguments: 扩展参数,用于扩展AMQP协议自制定化使用
-
队列声明
/**
* Declare a queue
* @see com.rabbitmq.client.AMQP.Queue.Declare
* @see com.rabbitmq.client.AMQP.Queue.DeclareOk
* @param queue the name of the queue
* @param durable true if we are declaring a durable queue (the queue will survive a server restart)
* @param exclusive true if we are declaring an exclusive queue (restricted to this connection)
* @param autoDelete true if we are declaring an autodelete queue (server will delete it when no longer in use)
* @param arguments other properties (construction arguments) for the queue
* @return a declaration-confirm method to indicate the queue was successfully declared
* @throws java.io.IOException if an error is encountered
*/
Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,
Map<String, Object> arguments) throws IOException;
方法 queueDeclare
的参数:第一个参数表示队列名称、第二个参数为是否持久化(true表示是,队列将在服务器重启时生存)、第三个参数为是否是独占队列(创建者可以使用的私有队列,断开后自动删除)、第四个参数为当所有消费者客户端连接断开时是否自动删除队列、第五个参数为队列的其他参数。
第三个参数(exclusive)
排他队列
如果你想创建一个只有自己可见的队列,即不允许其它用户访问,RabbitMQ允许你将一个Queue声明成为排他性的(Exclusive Queue)。该队列的特点是:
** 只对首次声明它的连接(Connection)可见
** 会在其连接断开的时候自动删除。
** 对于第一点,首先是强调首次声明,因为另外一个连接无法声明一个同样的排他性队列;其次是只区别连接(Connection)而不是通道(Channel),从同一个连接创建的不同的通道可以同时访问某一个排他性的队列。这里说的连接是指一个AMQPConnection,以RabbitMQ的Java客户端为例:
** 如果试图在一个不同的连接中重新声明或访问(如publish,consume)该排他性队列,会得到资源被锁定的错误:
**ESOURCE_LOCKED - cannot obtain exclusive access to locked queue 'UserLogin2'
** 对于第二点,RabbitMQ会自动删除这个队列,而不管这个队列是否被声明成持久性的(Durable =true)。 也就是说即使客户端程序将一个排他性的队列声明成了Durable的,只要调用了连接的Close方法或者客户端程序退出了,RabbitMQ都会删除这个队列。注意这里是连接断开的时候,而不是通道断开。这个其实前一点保持一致,只区别连接而非通道。
-
basicConsume的第二个参数autoAck: 应答模式,true:自动应答,即消费者获取到消息,该消息就会从队列中删除掉,false:手动应答,当从队列中取出消息后,需要程序员手动调用方法应答,如果没有应答,该消息还会再放进队列中,就会出现该消息一直没有被消费掉的现象。
-
这种简单队列的模式,系统会为每个队列隐式地绑定一个默认交换机,交换机名称为" (AMQP default)",类型为直连 direct,当你手动创建一个队列时,系统会自动将这个队列绑定到一个名称为空的 Direct 类型的交换机上,绑定的路由键 routing key 与队列名称相同,相当于
channel.queueBind(queue:"QUEUE_NAME", exchange:"(AMQP default)“, routingKey:"QUEUE_NAME");
虽然实例没有显式声明交换机,但是当路由键和队列名称一样时,就会将消息发送到这个默认的交换机中。这种方式比较简单,但是无法满足复杂的业务需求,所以通常在生产环境中很少使用这种方式。 -
The default exchange is implicitly bound to every queue, with a routing key equal to the queue name. It is not possible to explicitly bind to, or unbind from the default exchange. It also cannot be deleted.默认交换机隐式绑定到每个队列,其中路由键等于队列名称。不可能显式绑定到,或从缺省交换中解除绑定。它也不能被删除。
——引自 RabbitMQ 官方文档
-
声明交换机
/**
* Declare an exchange, via an interface that allows the complete set of
* arguments.
* @see com.rabbitmq.client.AMQP.Exchange.Declare
* @see com.rabbitmq.client.AMQP.Exchange.DeclareOk
* @param exchange the name of the exchange
* @param type the exchange type
* @param durable true if we are declaring a durable exchange (the exchange will survive a server restart)
* @param autoDelete true if the server should delete the exchange when it is no longer in use
* @param internal true if the exchange is internal, i.e. can't be directly
* published to by a client.
* @param arguments other properties (construction arguments) for the exchange
* @return a declaration-confirm method to indicate the exchange was successfully declared
* @throws java.io.IOException if an error is encountered
*/
Exchange.DeclareOk exchangeDeclare(String exchange,
String type,boolean durable,
boolean autoDelete,boolean internal,
Map<String, Object> arguments) throws IOException;
-
Name: 交换机名称
-
Type: 交换机类型direct、topic、 fanout、 headers
-
Durability: 是否需要持久化,true为持久化
-
Auto Delete: 当最后一个绑定到Exchange. 上的队列删除后,自动删除该Exchange
-
Internal: 当前Exchange是否用于RabbitMQ内部使用,默认为False
-
Arguments: 扩展参数,用于扩展AMQP协议自制定化使用
Spring Boot简单整合
参考《RabbitMQ(四):RabbitMQ与Spring Boot简单整合 快速尝鲜版》
Spring Boot 版与原生的写法相比不知道简便的多少。但是原生写法是基础,我们也是需要了解的。