diff options
Diffstat (limited to '2025/05/07')
| -rw-r--r-- | 2025/05/07/nestjs-bullmq-mail-business/index.html | 16 |
1 files changed, 8 insertions, 8 deletions
diff --git a/2025/05/07/nestjs-bullmq-mail-business/index.html b/2025/05/07/nestjs-bullmq-mail-business/index.html index 33c566c5..0315b2b9 100644 --- a/2025/05/07/nestjs-bullmq-mail-business/index.html +++ b/2025/05/07/nestjs-bullmq-mail-business/index.html @@ -196,14 +196,14 @@ <ol> <li>队列（在 NestJS 里由 <code>@Processor</code> 装饰的类）会被一个底层的 <code>Worker</code> 订阅。 </li> <li>有新任务（job）进来时，Worker 会调用写在该类里的 <code>async process(job: Job)</code> 方法。 </li> -<li>如果 <code>process()</code> 正常返回（即没有抛异常），Job 就被标记为 <strong>completed</strong>，然后才会去触发所有注册了 <code>@OnWorkerEvent('completed')</code> 的回调。</li> +<li>如果 <code>process()</code> 正常返回（即没有抛异常），Job 就被标记为 completed，然后才会去触发所有注册了 <code>@OnWorkerEvent('completed')</code> 的回调。</li> </ol> <p>也就是说：</p> <ul> -<li>**<code>process</code>**：是真正“干活”的地方，收到 job 之后立刻被调用，任何主业务逻辑（发邮件／写数据库／第三方请求等）都应该放这里。 </li> -<li><strong><code>onCompleted</code><strong>：只是一个事件监听器，</strong>在 job 已经成功完成之后</strong> 才会被触发，不会影响 job 的重试逻辑（也就是说，在这里抛错，job 已经算完成了，也不会重试）。</li> +<li><code>process</code>：是真正“干活”的地方，收到 job 之后立刻被调用，任何主业务逻辑（发邮件／写数据库／第三方请求等）都应该放这里。 </li> +<li><code>onCompleted</code>：只是一个事件监听器，在 job 已经成功完成之后 才会被触发，不会影响 job 的重试逻辑（也就是说，在这里抛错，job 已经算完成了，也不会重试）。</li> </ul> -<p>而我在此处的业务目的是 “用队列来做可靠的、可重试的邮件发送”，那么<strong>一定要把发送邮件的逻辑写到 <code>process()</code> 里</strong>，这样在 <code>commandBus.execute(new SendMailCommand(...))</code> 抛错时，BullMQ 会根据创建 JOB 时的重试策略（retry、backoff 等）自动重新入队。而把它放到 <code>onCompleted()</code>，只相当于 job 成功完成后的“事后通知”，一旦失败不会再重试，也无法利用 BullMQ 的锁、超时、重试机制。</p> +<p>而我在此处的业务目的是 “用队列来做可靠的、可重试的邮件发送”，那么一定要把发送邮件的逻辑写到 <code>process()</code> 里，这样在 <code>commandBus.execute(new SendMailCommand(...))</code> 抛错时，BullMQ 会根据创建 JOB 时的重试策略（retry、backoff 等）自动重新入队。而把它放到 <code>onCompleted()</code>，只相当于 job 成功完成后的“事后通知”，一旦失败不会再重试，也无法利用 BullMQ 的锁、超时、重试机制。</p> <p>举个最简化的调整示例，删掉 <code>onCompleted</code>，把真正的发信放到 <code>process</code>： </p> <figure class="highlight plaintext"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br></pre></td><td class="code"><pre><span class="line">// ... existing imports ...</span><br><span class="line"></span><br><span class="line">@Processor(process.env.MAILER_QUEUE_NAME || "gcpm-mailer")</span><br><span class="line">export class BullMQMailerProcesser extends WorkerHost {</span><br><span class="line"> constructor(</span><br><span class="line"> private readonly commandBus: CommandBus,</span><br><span class="line"> private readonly logger: LoggingService,</span><br><span class="line"> ) {</span><br><span class="line"> super();</span><br><span class="line"> }</span><br><span class="line"></span><br><span class="line"> // ① 当有新 job 拉取到时，这个方法会被调用</span><br><span class="line"> public async process(job: Job): Promise<void> {</span><br><span class="line"> const mailAggregate = new Mail(job.data.mail);</span><br><span class="line"> try {</span><br><span class="line"> await this.commandBus.execute(new SendMailCommand(mailAggregate));</span><br><span class="line"> } catch (err) {</span><br><span class="line"> this.logger.error(`邮件发送失败，jobId=${job.id}`, err);</span><br><span class="line"> // 抛出错误，触发重试或失败</span><br><span class="line"> throw err;</span><br><span class="line"> }</span><br><span class="line"> }</span><br><span class="line"></span><br><span class="line"> // ② onCompleted 仅在 process() 正常返回后触发，</span><br><span class="line"> // 不建议在这里执行核心业务（也无法触发重试）。</span><br><span class="line"> // @OnWorkerEvent("completed")</span><br><span class="line"> // async onCompleted(job: Job) { … }</span><br><span class="line">}</span><br></pre></td></tr></table></figure> @@ -213,9 +213,9 @@ <li>“Events → OnJobCompleted”：completed 事件只是一个监听钩子，不会参与重试。</li> </ul> <hr> -<p>而 <strong>重试次数本身并没有一个硬性上限</strong>，完全由添加 Job 时通过 <code>attempts</code> 这个选项来控制：</p> +<p>而 重试次数本身并没有一个硬性上限，完全由添加 Job 时通过 <code>attempts</code> 这个选项来控制：</p> <ul> -<li>默认情况下，如果不传 <code>attempts</code>（或不在 <code>defaultJobOptions</code> 里配置），Job <strong>不会自动重试</strong>（相当于 <code>attempts = 0</code>）。 </li> +<li>默认情况下，如果不传 <code>attempts</code>（或不在 <code>defaultJobOptions</code> 里配置），Job 不会自动重试（相当于 <code>attempts = 0</code>）。 </li> <li>如果在 <code>queue.add()</code>（或全局 <code>defaultJobOptions</code>）里设置了 <code>attempts: N</code>，那么 BullMQ 最多会让该 Job 运行 N 次（也就是初始执行 + N−1 次重试，或者根据文档含义最多触发 N 次失败） ，失败后才算真正移入失败集合。 </li> <li><code>attempts</code> 可以是任意的正整数（受 JavaScript <code>Number</code> 范围限制），BullMQ 本身不会再做额外的上限检查。</li> </ul> @@ -224,13 +224,13 @@ <hr> <p>还有一个需要注意的地方，在我的业务中，邮件发送的是一种时间区间报告，这个报告包含了过去二十四小时的一些系统中的事件，但如果重试有延迟策略或重试本身就有计算成本的话，这封邮件就不是 “过去二十四小时” 的了，因为重试带来了一个真空期。</p> -<p>换言之，这个问题本质上是——<strong>重试导致「发送时刻」与「原始 24 小时窗口」错开</strong>，从而让邮件里报出来的数据不再精确。常见的解决思路就是：<strong>把「窗口定义」或者「报表内容」在调度时就固化下来，真正的队列任务只负责发送</strong>，而不再实时去重新计算时间区间。</p> +<p>换言之，这个问题本质上是——重试导致「发送时刻」与「原始 24 小时窗口」错开，从而让邮件里报出来的数据不再精确。常见的解决思路就是：把「窗口定义」或者「报表内容」在调度时就固化下来，真正的队列任务只负责发送，而不再实时去重新计算时间区间。</p> <p>我想到了两种解决方案：</p> <p>一、任务参数里带上「时间区间」<br> 在 enqueue 的时候，就算出 windowStart/windowEnd，然后把它放到 <code>job.data</code> 里。无论后面 <code>process</code> 什么时候真正跑，都是基于同一个时间区间去查询：</p> <figure class="highlight ts"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">// 调度时</span></span><br><span class="line"><span class="keyword">const</span> now = <span class="keyword">new</span> <span class="title class_">Date</span>();</span><br><span class="line"><span class="keyword">const</span> windowStart = <span class="keyword">new</span> <span class="title class_">Date</span>(now.<span class="title function_">getTime</span>() - <span class="number">24</span> * <span class="number">60</span> * <span class="number">60</span> * <span class="number">1000</span>);</span><br><span class="line"><span class="keyword">await</span> <span class="variable language_">this</span>.<span class="property">mailerQueue</span>.<span class="title function_">add</span>(</span><br><span class="line"> id,</span><br><span class="line"> {</span><br><span class="line"> <span class="attr">mail</span>: <span class="keyword">new</span> <span class="title class_">Mail</span>({</span><br><span class="line"> ...options,</span><br><span class="line"> id,</span><br><span class="line"> <span class="attr">sentAt</span>: now,</span><br><span class="line"> <span class="attr">status</span>: <span class="title class_">MailStatus</span>.<span class="property">PENDING</span>,</span><br><span class="line"> windowStart,</span><br><span class="line"> <span class="attr">windowEnd</span>: now,</span><br><span class="line"> }),</span><br><span class="line"> },</span><br><span class="line"> {</span><br><span class="line"> <span class="attr">attempts</span>: <span class="number">3</span>,</span><br><span class="line"> <span class="attr">backoff</span>: { <span class="attr">type</span>: <span class="string">'exponential'</span>, <span class="attr">delay</span>: <span class="number">1000</span> },</span><br><span class="line"> },</span><br><span class="line">);</span><br><span class="line"></span><br><span class="line"><span class="comment">// process 里</span></span><br><span class="line"><span class="keyword">public</span> <span class="keyword">async</span> <span class="title function_">process</span>(<span class="params"><span class="attr">job</span>: <span class="title class_">Job</span></span>) {</span><br><span class="line"> <span class="keyword">const</span> { windowStart, windowEnd } = job.<span class="property">data</span>.<span class="property">mail</span>;</span><br><span class="line"> <span class="comment">// ① 只查询 [windowStart, windowEnd] 的事件</span></span><br><span class="line"> <span class="keyword">const</span> events = <span class="keyword">await</span> <span class="variable language_">this</span>.<span class="property">reportService</span>.<span class="title function_">findEvents</span>(windowStart, windowEnd);</span><br><span class="line"> <span class="keyword">const</span> reportHtml = <span class="keyword">await</span> <span class="variable language_">this</span>.<span class="property">reportService</span>.<span class="title function_">renderReport</span>(events);</span><br><span class="line"> <span class="keyword">await</span> <span class="variable language_">this</span>.<span class="property">commandBus</span>.<span class="title function_">execute</span>(<span class="keyword">new</span> <span class="title class_">SendMailCommand</span>(job.<span class="property">data</span>.<span class="property">mail</span>, reportHtml));</span><br><span class="line">}</span><br></pre></td></tr></table></figure> <p> ➜ 这样无是马上执行还是几次重试后才执行，数据规则都不会变。</p> <p>二、预先生成「静态报表内容」，挂到队列里<br> 如果计算成本很高，或者怕重复查询数据开销大，也可以在调度时就把最终的 HTML/Text/附件 都先打好，然后作为 <code>job.data</code> 传进去，真正的 <code>process()</code> 只做一次“发送”即可：<br> <figure class="highlight ts"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">// 调度时：先生成报告</span></span><br><span class="line"><span class="keyword">const</span> now = <span class="keyword">new</span> <span class="title class_">Date</span>();</span><br><span class="line"><span class="keyword">const</span> windowStart = <span class="keyword">new</span> <span class="title class_">Date</span>(now.<span class="title function_">getTime</span>() - <span class="number">24</span>*<span class="number">3600</span>*<span class="number">1000</span>);</span><br><span class="line"><span class="keyword">const</span> events = <span class="keyword">await</span> <span class="variable language_">this</span>.<span class="property">reportService</span>.<span class="title function_">findEvents</span>(windowStart, now);</span><br><span class="line"><span class="keyword">const</span> reportHtml = <span class="keyword">await</span> <span class="variable language_">this</span>.<span class="property">reportService</span>.<span class="title function_">renderReport</span>(events);</span><br><span class="line"></span><br><span class="line"><span class="comment">// 把静态内容塞到队列</span></span><br><span class="line"><span class="keyword">await</span> <span class="variable language_">this</span>.<span class="property">mailerQueue</span>.<span class="title function_">add</span>(</span><br><span class="line"> id,</span><br><span class="line"> {</span><br><span class="line"> <span class="attr">mail</span>: <span class="keyword">new</span> <span class="title class_">Mail</span>({ <span class="comment">/*…*/</span>, windowStart, <span class="attr">windowEnd</span>: now }),</span><br><span class="line"> reportHtml, <span class="comment">// <- 预渲染好的文本／HTML</span></span><br><span class="line"> <span class="attr">attachments</span>: […], <span class="comment">// <- 如果有附件也一并塞</span></span><br><span class="line"> },</span><br><span class="line"> { <span class="attr">attempts</span>: <span class="number">3</span>, <span class="attr">backoff</span>: { <span class="attr">type</span>: <span class="string">'fixed'</span>, <span class="attr">delay</span>: <span class="number">5_000</span> } },</span><br><span class="line">);</span><br><span class="line"></span><br><span class="line"><span class="comment">// process 里只关注发送</span></span><br><span class="line"><span class="keyword">public</span> <span class="keyword">async</span> <span class="title function_">process</span>(<span class="params"><span class="attr">job</span>: <span class="title class_">Job</span></span>) {</span><br><span class="line"> <span class="keyword">try</span> {</span><br><span class="line"> <span class="keyword">await</span> <span class="variable language_">this</span>.<span class="property">mailService</span>.<span class="title function_">send</span>({</span><br><span class="line"> <span class="attr">to</span>: job.<span class="property">data</span>.<span class="property">mail</span>.<span class="property">to</span>,</span><br><span class="line"> <span class="attr">subject</span>: <span class="string">`系统 24h 报表`</span>,</span><br><span class="line"> <span class="attr">html</span>: job.<span class="property">data</span>.<span class="property">reportHtml</span>,</span><br><span class="line"> <span class="attr">attachments</span>: job.<span class="property">data</span>.<span class="property">attachments</span>,</span><br><span class="line"> });</span><br><span class="line"> } <span class="keyword">catch</span> (e) {</span><br><span class="line"> <span class="keyword">throw</span> e; <span class="comment">// 触发重试</span></span><br><span class="line"> }</span><br><span class="line">}</span><br></pre></td></tr></table></figure><br> ➜ 重试带来的任何延迟，都不影响邮件正文，始终是一份「事先约定好、并且静态化」的报告。</p> -<p>这两种模式都能保证<strong>最终发送时的数据窗口</strong>或<strong>内容</strong>，与当初调度时的预期完全一致，不会因为重试延迟而出现“数据真空”或“多算/少算”问题。</p> +<p>这两种模式都能保证最终发送时的数据窗口或内容，与当初调度时的预期完全一致，不会因为重试延迟而出现“数据真空”或“多算/少算”问题。</p> <h2 id="参考文档："><a href="#参考文档：" class="headerlink" title="参考文档："></a>参考文档：</h2><ul> <li>“Retrying failing jobs” · BullMQ Guide<br><a target="_blank" rel="noopener" href="https://docs.bullmq.io/guide/retrying-failing-jobs">https://docs.bullmq.io/guide/retrying-failing-jobs</a></li> <li>BullMQ Guide & Patterns · Process Step Jobs (completed event only fires after process resolves)<br><a target="_blank" rel="noopener" href="https://docs.bullmq.io/patterns/process-step-jobs">https://docs.bullmq.io/patterns/process-step-jobs</a></li> |