RabbitMQ使用

本文将会讲解如何使用RabbitMQ实现延时重试和失败消息队列,实现可靠的消息消费,消费失败后,自动延时将消息重新投递,当达到一定的重试次数后,将消息投递到失败消息队列,等待人工介入处理。在这里我会带领大家一步一步的实现一个带有失败重试功能的发布订阅组件,使用该组件后可以非常简单的实现消息的发布订阅,在进行业务开发的时候,业务开发人员可以将主要精力放在业务逻辑实现上,而不需要花费时间去理解RabbitMQ的一些复杂概念。

概要

我们将会实现如下功能

  • 结合RabbitMQ的Topic模式和Work Queue模式实现生产方产生消息,消费方按需订阅,消息投递到消费方的队列之后,多个worker同时对消息进行消费
  • 结合RabbitMQ的 Message TTL 和 Dead Letter Exchange 实现消息的延时重试功能
  • 消息达到最大重试次数之后,将其投递到失败队列,等待人工介入处理bug后,重新将其加入队列消费
  • 具体流程见下图

xxx

  1. 生产者发布消息到主Exchange
  2. 主Exchange根据Routing Key将消息分发到对应的消息队列
  3. 多个消费者的worker进程同时对队列中的消息进行消费,因此它们之间采用“竞争”的方式来争取消息的消费
  4. 消息消费后,不管成功失败,都要返回ACK消费确认消息给队列,避免消息消费确认机制导致重复投递,同时,如果消息处理成功,则结束流程,否则进入重试阶段
  5. 如果重试次数小于设定的最大重试次数(3次),则将消息重新投递到Retry Exchange的重试队列
  6. 重试队列不需要消费者直接订阅,它会等待消息的有效时间过期之后,重新将消息投递给Dead Letter Exchange,我们在这里将其设置为主Exchange,实现延时后重新投递消息,这样消费者就可以重新消费消息
  7. 如果三次以上都是消费失败,则认为消息无法被处理,直接将消息投递给Failed Exchange的Failed Queue,这时候应用可以触发报警机制,以通知相关责任人处理
  8. 等待人工介入处理(解决bug)之后,重新将消息投递到主Exchange,这样就可以重新消费了

创建Exchange

为了实现消息的延时重试和失败存储,我们需要创建三个Exchange来处理消息。

  • master 主Exchange,发布消息时发布到该Exchange
  • master.retry 重试Exchange,消息处理失败时(3次以内),将消息重新投递给该Exchange
  • master.failed 失败Exchange,超过三次重试失败后,消息投递到该Exchange
  • 所有的Exchange声明(declare)必须使用以下参数
参数说明
exchange-Exchange名称
typetopicExchange类型
passivefalse如果Exchange已经存在,则返回成功,不存在则创建
durabletrue持久化存储Exchange,这里仅仅是Exchange本身持久化,消息和队列需要单独指定其持久化
no-waitfalse该方法需要应答确认
// 普通交换机
$this->channel->exchange_declare('master', 'topic', false, true, false);
// 重试交换机
$this->channel->exchange_declare('master.retry', 'topic', false, true, false);
// 失败交换机
$this->channel->exchange_declare('master.failed', 'topic', false, true, false);

在RabbitMQ的管理界面中,我们可以看到创建的三个Exchange
15261952974338

消息发布
消息发布时,使用basic_publish方法,参数如下

参数说明
message发布的消息对象
exchangemaster消息发布到的Exchange
routing-key路由KEY,用于标识消息类型
mandatoryfalse是否强制路由,指定了该选项后,如果没有订阅该消息,则会返回路由不可达错误
immediatefalse指定了当消息无法直接路由给消费者时如何处理

发布消息时,对于message对象,其内容建议使用json编码后的字符串,同时消息需要标识以下属性

'delivery_mode'=> 2 // 1为非持久化,2为持久化
   
$msg = new AMQPMessage($message->serialize(), [
    'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT,
]);
$this->channel->basic_publish($msg, 'master', $routingKey);

消息订阅

消息订阅的实现相对复杂一些,需要完成队列的声明以及队列和Exchange的绑定。

Declare Queue

对于每一个订阅消息的服务,都必须创建一个该服务对应的队列,将该队列绑定到关注的路由规则,这样之后,消息生产者将消息投递给Exchange之后,就会按照路由规则将消息分发到对应的队列供消费者消费了。

消费服务需要declare三个队列

  • [queue_name] 队列名称,格式符合 [服务名称]@订阅服务标识
  • [queue_name]@retry 重试队列
  • [queue_name]@failed 失败队列
订阅服务标识是客户端自己对订阅的分类标识符,比如用户中心服务(服务名称ucenter),包含两个订阅:user和enterprise,这里两个订阅的队列名称就为 ucenter@user和ucenter@enterprise,其对应的重试队列为 ucenter@user@retry和ucenter@enterprise@retry。

Declare队列时,参数规定规则如下

参数说明
queue队列名称
passivefalse队列不存在则创建,存在则直接成功
durabletrue队列持久化
exclusivefalse排他,指定该选项为true则队列只对当前连接有效,连接断开后自动删除
no-waitfalse该方法需要应答确认
auto-deletefalse当不再使用时,是否自动删除

对于@retry重试队列,需要指定额外参数

'x-dead-letter-exchange' => 'master'
'x-message-ttl'          => 30 * 1000 // 重试时间设置为30s
这里的两个header字段的含义是,在队列中延迟30s后,将该消息重新投递到x-dead-letter-exchange对应的Exchange中
$this->channel->queue_declare($queueName, false, true, false, false, false);
$this->channel->queue_declare($failedQueueName, false, true, false, false, false);
$this->channel->queue_declare(
    $retryQueueName, // 队列名称
    false,           // passive
    true,            // durable
    false,           // exclusive
    false,           // auto_delete
    false,           // nowait
    new AMQPTable([
        'x-dead-letter-exchange' => 'master',
        'x-message-ttl'          => 30 * 1000,
    ])
);

在RabbitMQ的管理界面中,Queues部分可以看到我们创建的三个队列

15261954013804

查看队列的详细信息,我们可以看到 queueName@retry 队列与其它两个队列的不同
15261955782834

Bind Exchange & Queue

创建完队列之后,需要将队列与Exchange绑定(bind),不同队列需要绑定到之前创建的对应的Exchange上面

Queue Exchange

[queue_name] master
[queue_name]@retry master.retry
[queue_name]@failed master.failed
绑定时,需要提供订阅的路由KEY,该路由KEY与消息发布时的路由KEY对应,区别是这里可以使用通配符同时订阅多种类型的消息。

参数 值 说明
queue – 绑定的队列
exchange – 绑定的Exchange
routing-key – 订阅的消息路由规则
no-wait false 该方法需要应答确认

$this->channel->queue_bind($queueName, 'master', $routingKey);
$this->channel->queue_bind($retryQueueName, 'master.retry', $routingKey);
$this->channel->queue_bind($failedQueueName, 'master.failed', $routingKey);

在RabbitMQ的管理界面中,我们可以看到该队列与Exchange和routing-key的绑定关系

15261958610725
15261958150612
15261958961191

消息消费实现

使用 basic_consume 对消息进行消费的时候,需要注意下面参数

参数 值 说明
queue – 消费的队列名称
consumer-tag – 消费者标识,留空即可
no_local false 如果设置了该字段,服务器将不会发布消息到 发布它的客户端
no_ack false 需要消费确认应答
exclusive false 排他访问,设置后只允许当前消费者访问该队列
nowait false 该方法需要应答确认
消费端在消费消息时,需要从消息中获取消息被消费的次数,以此判断该消息处理失败时重试还是发送到失败队列。

protected function getRetryCount(AMQPMessage $msg): int
{
    $retry = 0;
    if ($msg->has('application_headers')) {
        $headers = $msg->get('application_headers')->getNativeData();
        if (isset($headers['x-death'][0]['count'])) {
            $retry = $headers['x-death'][0]['count'];
        }
    }
    return (int)$retry;
}

消息消费完成后,需要发送消费确认消息给服务端,使用basic_ack方法

ack(delivery-tag=消息的delivery-tag标识)
$this->channel->basic_consume(
    $queueName,
    '',    // customer_tag
    false, // no_local
    false, // no_ack
    false, // exclusive
    false, // nowait
    function (AMQPMessage $msg) use ($queueName, $routingKey, $callback) {
        ...
        $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
    }
);

如果消息处理中出现异常,应该将该消息重新投递到重试Exchange,等待下次重试

basic_publish(msg, 'master.retry', routing-key)
ack(delivery-tag) // 不要忘记了应答消费成功消息

如果判断重试次数大于3次,仍然处理失败,则应该讲消息投递到失败Exchange,等待人工处理

basic_publish(msg, 'master.failed', routing-key)
ack(delivery-tag) // 不要忘记了应答消费成功消息
一定不要忘记ack消息,因为重试、失败都是通过将消息重新投递到重试、失败Exchange来实现的,如果忘记ack,则该消息在超时或者连接断开后,会重新被重新投递给消费者,如果消费者依旧无法处理,则会造成死循环。

Java代码

try {

String message = new String(body, "UTF-8");
// 消息处理函数
handler.handle(message, envelope.getRoutingKey());

} catch (Exception e) {

long retryCount = getRetryCount(properties);
if (retryCount > 3) {
    // 重试次数大于3次,则自动加入到失败队列
    channel.basicPublish("master.failed", envelope.getRoutingKey(), MessageProperties.PERSISTENT_BASIC, body);
} else {
    // 重试次数小于3,则加入到重试队列,30s后再重试
    channel.basicPublish("master.retry", envelope.getRoutingKey(), properties, body);
}

}
失败任务重试
如果任务重试三次仍未成功,则会被投递到失败队列,这时候需要人工处理程序异常,处理完毕后,需要将消息重新投递到队列进行处理,这里唯一需要做的就是从失败队列订阅消息,然后获取到消息后,清空其application_headers头信息,然后重新投递到master这个Exchange即可。

Java代码

channel.basicPublish(

'master', 
envelope.getRoutingKey(),
MessageProperties.PERSISTENT_BASIC,
body

);
PHP代码

$msg->set('application_headers', new AMQPTable([]));
$this->channel->basic_publish(

$msg,
'master',
$msg->get('routing_key')

);
怎么使用
队列和Exchange以及发布订阅的关系我们就说完了,那么使用起来是什么效果呢?这里我们以Java代码为例

// 发布消息
Publisher publisher = new Publisher(factory.newConnection(), 'master');
publisher.publish("{"id":121, "name":"guanyiyao"}", "user.create");
// 订阅消息
new Subscriber(factory.newConnection(), Main.EXCHANGE_NAME)

.init("user-monitor", "user.*")
.subscribe((message, routingKey) -> {
    // TODO 业务逻辑
    System.out.printf("    <%s> message consumed: %s\n", routingKey, message);
}

);
总结
使用RabbitMQ时,实现延时重试和失败队列的方式并不仅仅局限于本文中描述的方法,如果读者有更好的实现方案,欢迎拍砖,在这里我也只是抛砖引玉了。本文中讲述的方法还有很多优化空间,读者也可以试着去改进其实现方案,比如本文中使用了三个Exchagne,是否只使用一个Exchange也能实现本文中所讲述的功能。

原文链接----点此