本节继续介绍RabbitMQ的进阶使用,生产者确认机制和消费端的要点.最后将结合之前所学阐述RabbitMQ的全局消息保障级别和实现.
1. 生产者确认机制
至此为止，我们学习了RabbitMQ提供的消息持久化机制、消费者确认机制。那如何保证消息正确的发给了RabbitMQ？针对这个问题，RabbitMQ提供了两种解决方式：
- 通过事务机制实现
- 通过发送方确认机制实现
- 事务机制
RabbitMQ客户端中与事务相关的方法有三个:channel.txSelect、channel.txCommit和channel.txRollback
这里提供一下代码实例:
public void txTest() throws IOException {
Connection connection = ConnectionCreator.getConnection();
Channel channel = connection.createChannel();
try {
channel.txSelect();
channel.basicPublish(RabbitMQUtils.TEST_EXCHANGE, RabbitMQUtils.TEST_ROUTING, MessageProperties.TEXT_PLAIN, "transaction test".getBytes());
//int i = 1 / 0;
channel.txCommit();
} catch (Exception e) {
channel.txRollback();
}
RabbitMQUtils.close(channel, connection);
}
事务机制在一条消息发送之后会使发送端阻塞,等待RabbitMQ的回应,之后才发送下一条消息.可以预见的QPS将会非常低.
- 发送发确认机制: publisher confirm
RabbitMQ通过channel.confirmSelect() 方法开启确认模式,确认模式有同步和异步两种方式。
- 同步确认：
RabbitMQ提供了四种同步确认的方式：
/**
* Wait until all messages published since the last call have been
* either ack'd or nack'd by the broker. Note, when called on a
* non-Confirm channel, waitForConfirms throws an IllegalStateException.
* @return whether all the messages were ack'd (and none were nack'd)
* @throws java.lang.IllegalStateException
*/
boolean waitForConfirms() throws InterruptedException;
/**
* Wait until all messages published since the last call have been
* either ack'd or nack'd by the broker; or until timeout elapses.
* If the timeout expires a TimeoutException is thrown. When
* called on a non-Confirm channel, waitForConfirms throws an
* IllegalStateException.
* @return whether all the messages were ack'd (and none were nack'd)
* @throws java.lang.IllegalStateException
*/
boolean waitForConfirms(long timeout) throws InterruptedException, TimeoutException;
/** Wait until all messages published since the last call have
* been either ack'd or nack'd by the broker. If any of the
* messages were nack'd, waitForConfirmsOrDie will throw an
* IOException. When called on a non-Confirm channel, it will
* throw an IllegalStateException.
* @throws java.lang.IllegalStateException
*/
void waitForConfirmsOrDie() throws IOException, InterruptedException;
/** Wait until all messages published since the last call have
* been either ack'd or nack'd by the broker; or until timeout elapses.
* If the timeout expires a TimeoutException is thrown. If any of the
* messages were nack'd, waitForConfirmsOrDie will throw an
* IOException. When called on a non-Confirm channel, it will
* throw an IllegalStateException.
* @throws java.lang.IllegalStateException
*/
void waitForConfirmsOrDie(long timeout) throws IOException, InterruptedException, TimeoutException;
如果信道没有开启publisher confirm模式,调用任何waitForConfirms方法都会报IlleaglStateException.对于没有参数的waitForConfirms()方法,其返回的条件是收到服务端的basic.ack或basic.nack或者被中断.对于 boolean waitForConfirms(long timeout)方法,回应超时后悔抛出TimeoutException.对于两个waitForConfirmsOrDie方法,在收到服务端的basic.nack之后,会抛出IOException.业务代码可以根据自身特性灵活处理,保障消息的可靠发送.
下面是一个demo:
public void syncConfirm() throws IOException, InterruptedException {
Connection connection = ConnectionCreator.getConnection();
Channel channel = connection.createChannel();
channel.confirmSelect();
channel.basicPublish(RabbitMQUtils.TEST_EXCHANGE, RabbitMQUtils.TEST_ROUTING, null, "confirm message".getBytes());
if (!channel.waitForConfirms()) {
System.out.println("未收到服务端确认消息");
}
RabbitMQUtils.close(channel, connection);
}
实际上同步确认机制并不会比事务机制提高太多的QPS,两者都需要等待消息确认写入后才会返回.在同步等待的方式下,发送一条消息的交互命令是两条:basic.publish和basic.ack.事务机制是三条:basic.publish,tx.commit,commit-ok.
- 异步确认
RabbitMQ提供异步确认机制,在客户端的channel中添加ConfirmListener这个回调接口.这个接口提供了handleAck和handleNack方法,分别用于处理消息确认和消息拒绝方法.每个方法都包含一个deliveryTag,作为消息的唯一有序序号.
这里依然提供一个demo
public void asyncConfirm() throws Exception {
Connection connection = ConnectionCreator.getConnection();
Channel channel = connection.createChannel();
channel.confirmSelect();
channel.addConfirmListener(new ConfirmListener() {
@Override
public void handleAck(long deliveryTag, boolean multiple) throws IOException {
synchronized (this) {
count++;
}
System.out.println("收到消息编号为: " + deliveryTag + "的回调通知");
if (multiple) {
set.headSet(deliveryTag + 1).clear();
} else {
set.remove(deliveryTag);
}
}
@Override
public void handleNack(long deliveryTag, boolean multiple) throws IOException {
System.out.println("消息编号为" + deliveryTag + "的消息被拒绝");
if (multiple) {
//todo 批量处理重发逻辑
} else {
// todo 处理当前消息的重发逻辑
}
}
});
int n = 500;
while (n > 0) {
Thread.sleep(10);
n--;
long next = channel.getNextPublishSeqNo();
channel.basicPublish(RabbitMQUtils.TEST_EXCHANGE, RabbitMQUtils.TEST_ROUTING, MessageProperties.TEXT_PLAIN,
("async message " + n).getBytes());
set.add(next);
}
}
2. 消费端要点
对于RabbitMQ的客户端,有以下几点需要注意:
- 消息分发
当RabbitMQ有多个消费者时,队列收到的消息将以轮询的方式分发给消费者.然而轮询是完全平均的分发机制.假如某些消费者任务繁重,或者机器性能优越能够拥有更高的吞吐率.需要按某些规则分发消息.
这时可以用到channel.basicQoS(int preretchCount),该方法允许限制信道上的消费者能保持的最大未确认消息数量.这种机制可以类比TCP/IP中的滑动窗口.需要注意的是basic.Qos对拉模式的消费者无效 - 消息的顺序性
RabbitMQ不能保证消息的顺序发送和顺序消费.可以通过业务逻辑,比如消息体内添加全局有序标识.
3. 消息传输保障
一般来说,消息中间件的传输保障分为三个层级:
- at most once: 最多一次.消息可能会丢失,但不会重复传输
- at least once: 最少一次. 消息绝不会丢失,但可能重复
- excatly once: 恰好一次.每条消息肯定会被传输一次并且仅传输一次
RabbitMQ可以支持其中的最多一次和最少一次.
即可以保证每条消息至少被发送和消费一次.实现这种投递需要以下几个方面:
- 消息生产者需要开启事务或者publisher confirem机制,并增加重发逻辑.保证消息至少能成功发送到RabbitMQ一次;
- 消息生产者需要配合使用mandatory参数或者备份交换器来确保消息能路由到某个队列中.进而使消息能够保存而不会被丢弃
- 消息和队列都要进行持久化处理.确保RabbitMQ服务器在遇到异常时不会造成消息丢失
- 消费者需要将autoAck设置为false,然后手动确认消息已被正确消费.
