1. 死信队列

"死信"模式 指的是,当消费者不能处理接收到的消息时,将这个消息重新发布到另外一个队列中,等待重试或者人工干预。

  • 消息被拒绝 (basic.reject or basic.nack) 且带 requeue=false 参数
  • 消息的TTL-存活时间已经过期
  • 队列长度限制被超越(队列满)

x-max-length :限制队列的最大长度 x-dead-letter-exchange:出现dead letter之后将dead letter重新发送到指定exchange x-dead-letter-routing-key:出现dead letter之后将dead letter重新按照指定的routing-key发送

2. 延迟消息使用

通过死信队列实现

2.1. 形成条件:

  • 消息被丢弃 ? requeue=false (reject 并且不删除消息)
  • 消息的ttl 到期
  • 超过队列设置长度

2.2. 设置队列参数

Optional name of an exchange to which messages will be republished if they are rejected or expire.
实际上,官方用词 "rejected" 应该被翻译成 : "丢弃"或者"抛弃",而不是"拒绝",也就是说,这里的 rejected 和 expire 是队列里面的消息的状态.而不是队列的动作.
  • 🌟 - x-overflow:String 支持参数 drop-head 默认丢弃 /reject-publish 拒绝消息 请使用 默认参数

  • x-dead-letter-exchange --> 重新发送到指定exchange

  • x-dead-letter-routing-key --> 按照指定的routing-key发送

2.3. 发送消息

  • x-message-ttl:Number ,发布的消息在队列中存在多长时间后被取消(单位毫秒) 可以对单个消息设置过期时间
# 使用 bschmitt/laravel-amqp

\Amqp::publish('routing-key', 'message' , [
      'vhost'=>'/test',
    'exchange_type' => 'direct',
    'exchange' => 'test.direct',
    'exchange_durable'=>true, //交换机 持久化
    'exchange_internal'=>false, //内部使用
    'exchange_auto_delete'  => false,
    'exchange_properties'=>[

    ],
    'queue_force_declare' => true, //强制持久化 队列持久 的话 交换机也要持久化 否则无法连接
    'queue_exclusive' => true, // 排他性 只对当前连接可见
    'queue_auto_delete'     => false,
    'queue' => 'test_queue_name',
    'queue_properties'=>[
        'x-message-ttl' => ['I',30000], //I 看上一片下面类型介绍 30s 过期
        'x-dead-letter-exchange'=>'ttt.exchange',
        'x-dead-letter-routing-key'=>'ttt_bk'
    ]
]);

2.4. 接收消息

ACK && REJECT
    /**
     * Acknowledges a message
     *
     * @param AMQPMessage $message
     */
    public function acknowledge(AMQPMessage $message)
    {
        $message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']);
        if ($message->body === 'quit') {
            $message->delivery_info['channel']->basic_cancel($message->delivery_info['consumer_tag']);
        }
    }
    /**
     * Rejects a message and requeues it if wanted (default: false)
     *
     * @param AMQPMessage $message
     * @param bool    $requeue
     */
    public function reject(AMQPMessage $message, $requeue = false)
    {
        $message->delivery_info['channel']->basic_reject($message->delivery_info['delivery_tag'], $requeue);
    }

-------------------------------------------------
\Amqp::consume('ttt_bk', function ($message, $resolver) {
    var_dump($message->body);
    $resolver->acknowledge($message); //ack
    # $resolver->reject($message,true|false); // Rejects a message and requeues it if wanted (default: false)  拒绝消息
}, [
    'exchange' => 'ttt.exchange',
    'exchange_type' => 'fanout',
    'queue_force_declare' => true,
     .... 参数和发布的一样 省略。。。。
    'persistent' => true
]);

然后 消费者 指定 x-dead-letter-exchange 和 routing-key 进行消费处理

results matching ""

    No results matching ""