nodejs+rabbitmq实现任务延迟执行与重试控制

男娘i 2022-10-19 00:43 202阅读 0赞

> 以发送邮件任务为例，第一次发送邮件失败后，延时30s后继续发送此邮件，如果如此重试3次后仍然失败，则停止发送。

![在这里插入图片描述][20210708170258601.png]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NDUzMjY2_size_16_color_FFFFFF_t_70]

*  send\_email: 发送邮件任务队列(**死信队列**)
 *  send\_email\_retry: 发送邮件任务重试队列
 *  send\_email\_fail: 发送邮件任务失败队列

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NDUzMjY2_size_16_color_FFFFFF_t_70 1]

### 生产者 ###

1.  **创建send\_email队列，并给队列绑定交换机`exchange`和路由`routingkey`**

await ch.assertQueue(queue_send_email, { 
            durable: true
        })
    await ch.assertExchange(dlx_routing_exchange)
    await ch.bindQueue(queue_send_email,dlx_routing_exchange, dlx_routing_key)

1.  **创建queue\_retry队列，指定死信交换机`deadLetterExchange`和死信路由`deadLetterRoutingKey`以及队列消息过期时间`messageTtl`**  
    可以发现我们指定的queue\_retry队列的死信交换机和死信路由和queue\_email队列声明的交换机和路由一致。这样只要queue\_retry队列中的消息过期后，就会自动从queue\_retry队列转移到queue\_email队列,而queue\_email的消费者又可以消费了，这样就实现了延迟消费。而被死信交换机和死信路由绑定的队列(queue\_email)就叫做 **死信队列**。

await ch.assertQueue(queue_retry, { 
            durable: true,
            deadLetterExchange: dlx_routing_exchange,
            deadLetterRoutingKey: dlx_routing_key,
            messageTtl: 30 * 1000
        })

1.  **创建queue\_fail队列**  
    失败队列主要就是存储重试3次后仍然失败的发邮件任务，供人工查看。

await ch.assertQueue(queue_fail, { 
            durable: true
        })

1.  **生产者创建发送邮件任务到queue\_email队列**  
    注意: 在headers中指定了一个count参数，用来标识重试的次数

await ch.sendToQueue(queue, Buffer.from('发送邮件'), { 
            deliveryMode: true,
            headers: { 
                count
            }
        })

### 消费者 ###

await ch.consume(queue, (msg) => { 
             //获取重试的次数
            let retryCount = msg.properties.headers.count
            const result = sendEmail();
            if (result) {  //发送成功
                console.log('发送成功')
            }
            if (!result && retryCount < 3) {  //发送失败重试，且重试次数小于3
                //将消息发送到重试队列，并且重试次数+1
                sendToWorker(queue_retry, ++retryCount)
            } else { 
                //将消息发送到失败队列
                sendToWorker(queue_fail, retryCount)
            }
            //不管是否发送成功能都要ack
            ch.ack(msg)
        }, { 
            noAck: false
        })

### 完整代码 ###

[20210708170258601.png]: /images/20221005/a35357d6dd97425dbb01f87f7fd03460.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NDUzMjY2_size_16_color_FFFFFF_t_70]: /images/20221005/4c3763f7caee4d9b85061b6de37683d6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NDUzMjY2_size_16_color_FFFFFF_t_70 1]: /images/20221005/aa4c73ffaac24097974346083a05d4fa.png