内容简介：实现RabbitMQ的消费者有两种模式，推模式（Push）和拉模式（Pull）。实现推模式推荐的方式是继承推模式是最常用的，但是有些情况下推模式并不适用的，比如说：
实现RabbitMQ的消费者有两种模式，推模式（Push）和拉模式（Pull）。
实现推模式推荐的方式是继承 DefaultConsumer
基类，也可以使用Spring AMQP的 SimpleMessageListenerContainer
。
推模式是最常用的，但是有些情况下推模式并不适用的，比如说：
- 由于某些限制，消费者在某个条件成立时才能消费消息
- 需要批量拉取消息进行处理
实现拉模式
RabbitMQ的Channel提供了 basicGet
方法用于拉取消息。
/**
* Retrieve a message from a queue using {@link com.rabbitmq.client.AMQP.Basic.Get}
* @see com.rabbitmq.client.AMQP.Basic.Get
* @see com.rabbitmq.client.AMQP.Basic.GetOk
* @see com.rabbitmq.client.AMQP.Basic.GetEmpty
* @param queue the name of the queue
* @param autoAck true if the server should consider messages
* acknowledged once delivered; false if the server should expect
* explicit acknowledgements
* @return a {@link GetResponse} containing the retrieved message data
* @throws java.io.IOException if an error is encountered
*/
GetResponse basicGet(String queue, boolean autoAck) throws IOException;
basicGet
返回 GetResponse
类。
public class GetResponse {
private final Envelope envelope;
private final BasicProperties props;
private final byte[] body;
private final int messageCount;
// ...
rabbitmq-client版本4.0.3
使用 basicGet
拉取消息需要注意：
basicGet DefaultConsumer
示例代码：
private void consume(Channel channel) throws IOException, InterruptedException {
while (true) {
if (!isConditionSatisfied()) {
TimeUnit.MILLISECONDS.sleep(1);
continue;
}
GetResponse response = channel.basicGet(CAOSH_TEST_QUEUE, false);
if (response == null) {
TimeUnit.MILLISECONDS.sleep(1);
continue;
}
String data = new String(response.getBody());
logger.info("Get message <= {}", data);
channel.basicAck(response.getEnvelope().getDeliveryTag(), false);
}
}
批量拉取消息
RabbitMQ支持客户端批量拉取消息，客户端可以连续调用 basicGet
方法拉取多条消息，处理完成之后一次性ACK。需要注意：
basicGet basicAck
示例代码：
String bridgeQueueName = extractorProperties.getBridgeQueueName();
int batchSize = extractorProperties.getBatchSize();
List<GetResponse> responseList = Lists.newArrayListWithCapacity(batchSize);
long tag = 0;
while (responseList.size() < batchSize) {
GetResponse getResponse = channel.basicGet(bridgeQueueName, false);
if (getResponse == null) {
break;
}
responseList.add(getResponse);
tag = getResponse.getEnvelope().getDeliveryTag();
}
if (responseList.isEmpty()) {
TimeUnit.MILLISECONDS.sleep(1);
} else {
logger.info("Get <{}> responses this batch", responseList.size());
// handle messages
channel.basicAck(tag, true);
}
关于QueueingConsumer
QueueingConsumer
在客户端本地使用 BlockingQueue
缓冲消息，其nextDelivery方法也可以用于实现拉模式（其本质上是 BlockingQueue.take
），但是 QueueingConsumer
现在已经标记为Deprecated。
根据 官方文档
的解释， QueueingConsumer
有两个坑：
Firstly, the Consumer could stall the processing of all Channels on the Connection. Secondly, if a Consumer made a recursive synchronous call into its Channel the client would deadlock.
建议不要使用 QueueingConsumer
。
以上所述就是小编给大家介绍的《RabbitMQ拉模式批量消费消息》，希望对大家有所帮助，如果大家有任何疑问请给我留言，小编会及时回复大家的。在此也非常感谢大家对 码农网 的支持！
猜你喜欢:
- 玩转KCP(2)-流模式和消息模式
- RabbitMQ消息交换模式简介
- 消息中间件的四种投递模式对比
- 重学 Java 设计模式：实战观察者模式「模拟类似小客车指标摇号过程，监听消息通知用户中签场景」
- WebSocket的故事（二）—— Spring中如何利用STOMP快速构建WebSocket广播式消息模式
- WebSocket的故事（三）—— Springboot中，如何利用WebSocket和STOMP快速构建点对点的消息模式（1）
本站部分资源来源于网络，本站转载出于传递更多信息之目的，版权归原作者或者来源机构所有，如转载稿涉及版权问题，请联系我们。
