summaryrefslogtreecommitdiff
path: root/2025/05/07/nestjs-bullmq-mail-business/index.html
blob: 10baaa9523212afd4b4c45570156b360861b1ecb (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
<!DOCTYPE html>
<html lang="en">
  <head>
    <meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0, maximum-scale=1.0, minimum-scale=1.0">
<meta http-equiv="X-UA-Compatible" content="ie=edge">

    <meta name="author" content="韩暮秋">
    
        
            <meta name="subtitle" content="暮秋小屋">
            
                
                    <meta name="description" content="这里是暮秋小屋,思念和灵感的寄存处">
                    
                        
                            <meta name="keywords" content="韩暮秋,MuqiuHan,'Muqiu Han', 'muqiu han', muqiuhan">
                            

                                

                                    <title>
                                        
                                            NestJS bullmq 邮件发送业务中的小 tips | 
                                                    暮秋小屋
                                    </title>

                                    
                                        
                                            <link rel="icon" href="/favicon.ico">
                                            

                                                        
                                                            
                                                                <!-- stylesheets list from _config.yml -->
                                                                
                                                                    <link rel="stylesheet" href="/css/style.css">
                                                                    
                                                                        

            
                <link rel="preload" href="/fonts/TensentypeCode.woff2" as="font" type="font/woff2" crossorigin>

                                                                    
                                                                        
                                                                            <!-- scripts list from _config.yml -->
                                                                            
                                                                                <script
                                                                                    src="/js/menu.js"></script>
                                                                                
                                                                                    

                                                                                        
                                                                                            
                                                                                                <script
                                                                                                    src="https://polyfill.alicdn.com/polyfill.js?features=es6"></script>
                                                                                                <script
                                                                                                    id="MathJax-script"
                                                                                                    async
                                                                                                    src="https://lf6-cdn-tos.bytecdntp.com/cdn/expire-1-M/mathjax/3.2.0/es5/tex-mml-chtml.js"></script>
                                                                                                

                                                                                                    
                                                                                                        
  <meta name="generator" content="Hexo 6.3.0"></head>
  <body>
    <div class="wrapper">

      <div class="header">
  <div class="flex-container">
    <div class="header-inner">
      <div class="site-brand-container">
        <a href="/">
          
              暮秋小屋
                
        </a>
      </div>
      <div id="menu-btn" class="menu-btn" onclick="toggleMenu()">
        菜单
      </div>
      <nav class="site-nav">
        <ul class="menu-list">
          
            
                <li class="menu-item">
                  <a href="/">
                    主页
                  </a>
                </li>
                
                  
            
                <li class="menu-item">
                  <a href="/categories/gallery/">
                    日记本
                  </a>
                </li>
                
                  
            
                <li class="menu-item">
                  <a href="/tags/Medicine/">
                    泛医学
                  </a>
                </li>
                
                  
            
                <li class="menu-item">
                  <a href="/tags/Technique/">
                    计算机
                  </a>
                </li>
                
                  
            
                <li class="menu-item">
                  <a href="/tags/Life/">
                    生活
                  </a>
                </li>
                
                  
            
                <li class="menu-item">
                  <a href="/archives">
                    全部
                  </a>
                </li>
                
                  
            
                <li class="menu-item">
                  <a href="/about">
                    关于
                  </a>
                </li>
                
                  
                    
                      <li class="menu-item">
                        <a href="/search">搜索</a>
                      </li>
                      
        </ul>
      </nav>
    </div>
  </div>
</div>


      <div class="main">
        <div class="flex-container">
          <article id="post">

  
    <div class="post-head">
    <div class="post-info">
        <div class="tag-list">
            
                
                    <span class="post-tag">
                        <a href="/tags/Technique/">
                            Technique
                        </a>
                    </span>    
                           
            
        </div>
        <div class="post-title">
            
            
                NestJS bullmq 邮件发送业务中的小 tips
            
            
        </div>
        <span class="post-date">
            May 7, 2025
        </span>
    </div>
    <div class="post-img">
        
            <div class="h-line-primary"></div>
              
    </div>
</div>
    <div class="post-content">
    <p>在 BullMQ(以及它在 NestJS 里包装的 <code>@Processor</code>/<code>WorkerHost</code>)里,整个生命周期大致是这样的:</p>
<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 就被标记为 completed,然后才会去触发所有注册了 <code>@OnWorkerEvent('completed')</code> 的回调。</li>
</ol>
<p>也就是说:</p>
<ul>
<li><code>process</code>:是真正“干活”的地方,收到 job 之后立刻被调用,任何主业务逻辑(发邮件/写数据库/第三方请求等)都应该放这里。</li>
<li><code>onCompleted</code>:只是一个事件监听器,在 job 已经成功完成之后 才会被触发,不会影响 job 的重试逻辑(也就是说,在这里抛错,job 已经算完成了,也不会重试)。</li>
</ul>
<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 || &quot;gcpm-mailer&quot;)</span><br><span class="line">export class BullMQMailerProcesser extends WorkerHost &#123;</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">  ) &#123;</span><br><span class="line">    super();</span><br><span class="line">  &#125;</span><br><span class="line"></span><br><span class="line">  // ① 当有新 job 拉取到时,这个方法会被调用</span><br><span class="line">  public async process(job: Job): Promise&lt;void&gt; &#123;</span><br><span class="line">    const mailAggregate = new Mail(job.data.mail);</span><br><span class="line">    try &#123;</span><br><span class="line">      await this.commandBus.execute(new SendMailCommand(mailAggregate));</span><br><span class="line">    &#125; catch (err) &#123;</span><br><span class="line">      this.logger.error(`邮件发送失败,jobId=$&#123;job.id&#125;`, err);</span><br><span class="line">      // 抛出错误,触发重试或失败</span><br><span class="line">      throw err;</span><br><span class="line">    &#125;</span><br><span class="line">  &#125;</span><br><span class="line"></span><br><span class="line">  // ② onCompleted 仅在 process() 正常返回后触发,</span><br><span class="line">  //    不建议在这里执行核心业务(也无法触发重试)。</span><br><span class="line">  // @OnWorkerEvent(&quot;completed&quot;)</span><br><span class="line">  // async onCompleted(job: Job) &#123; … &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>参考 BullMQ 官方文档:</p>
<ul>
<li>“Workers → Sandboxed processors”:Worker 拉到 job 就调用注册的处理函数,然后根据返回/抛错把 job 标记成 completed 或 failed。</li>
<li>“Events → OnJobCompleted”:completed 事件只是一个监听钩子,不会参与重试。</li>
</ul>
<hr>
<p>而 重试次数本身并没有一个硬性上限,完全由添加 Job 时通过 <code>attempts</code> 这个选项来控制:</p>
<ul>
<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>
<p>示例(给某封邮件最多重试 3 次):</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></pre></td><td class="code"><pre><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">  &#123; <span class="attr">mail</span>: <span class="keyword">new</span> <span class="title class_">Mail</span>(<span class="comment">/*…*/</span> ) &#125;,</span><br><span class="line">  &#123;</span><br><span class="line">    <span class="attr">attempts</span>: <span class="number">3</span>,                  <span class="comment">// 最多尝试 3 次</span></span><br><span class="line">    <span class="attr">backoff</span>: &#123;                    <span class="comment">// 重试时的延迟策略(可选)</span></span><br><span class="line">      <span class="attr">type</span>: <span class="string">&#x27;exponential&#x27;</span>,</span><br><span class="line">      <span class="attr">delay</span>: <span class="number">1000</span>,</span><br><span class="line">    &#125;,</span><br><span class="line">  &#125;,</span><br><span class="line">);</span><br></pre></td></tr></table></figure>
<hr>
<p>还有一个需要注意的地方,在我的业务中,邮件发送的是一种时间区间报告,这个报告包含了过去二十四小时的一些系统中的事件,但如果重试有延迟策略或重试本身就有计算成本的话,这封邮件就不是 “过去二十四小时” 的了,因为重试带来了一个真空期。</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">  &#123;</span><br><span class="line">    <span class="attr">mail</span>: <span class="keyword">new</span> <span class="title class_">Mail</span>(&#123;</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">    &#125;),</span><br><span class="line">  &#125;,</span><br><span class="line">  &#123;</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>: &#123; <span class="attr">type</span>: <span class="string">&#x27;exponential&#x27;</span>, <span class="attr">delay</span>: <span class="number">1000</span> &#125;,</span><br><span class="line">  &#125;,</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>) &#123;</span><br><span class="line">  <span class="keyword">const</span> &#123; windowStart, windowEnd &#125; = 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">&#125;</span><br></pre></td></tr></table></figure>
<p>➜ 这样无是马上执行还是几次重试后才执行,数据规则都不会变。</p>
<p>二、预先生成「静态报表内容」,挂到队列里<br>
如果计算成本很高,或者怕重复查询数据开销大,也可以在调度时就把最终的 HTML/Text/附件 都先打好,然后作为 <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><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">  &#123;</span><br><span class="line">    <span class="attr">mail</span>: <span class="keyword">new</span> <span class="title class_">Mail</span>(&#123; <span class="comment">/*…*/</span>, windowStart, <span class="attr">windowEnd</span>: now &#125;),</span><br><span class="line">    reportHtml,       <span class="comment">// &lt;- 预渲染好的文本/HTML</span></span><br><span class="line">    <span class="attr">attachments</span>: […], <span class="comment">// &lt;- 如果有附件也一并塞</span></span><br><span class="line">  &#125;,</span><br><span class="line">  &#123; <span class="attr">attempts</span>: <span class="number">3</span>, <span class="attr">backoff</span>: &#123; <span class="attr">type</span>: <span class="string">&#x27;fixed&#x27;</span>, <span class="attr">delay</span>: <span class="number">5_000</span> &#125; &#125;,</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>) &#123;</span><br><span class="line">  <span class="keyword">try</span> &#123;</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>(&#123;</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">    &#125;);</span><br><span class="line">  &#125; <span class="keyword">catch</span> (e) &#123;</span><br><span class="line">    <span class="keyword">throw</span> e; <span class="comment">// 触发重试</span></span><br><span class="line">  &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>➜ 重试带来的任何延迟,都不影响邮件正文,始终是一份「事先约定好、并且静态化」的报告。</p>
<p>这两种模式都能保证最终发送时的数据窗口或内容,与当初调度时的预期完全一致,不会因为重试延迟而出现“数据真空”或“多算/少算”问题。</p>
<h2 id="参考文档:"><a class="header-anchor" href="#参考文档:">¶</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 &amp; 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>
</ul>

</div> 

<script>
    window.onload = detectors();
</script>
    <div class="post-footer">
    <div class="h-line-primary"></div>
    <nav class="post-nav">
        <div class="prev-item">
           
                <div class="icon arrow-left"></div>
                <div class="post-link">
                    <a href="/2025/05/08/Multiplayer-Collaborative-Systems-tips/">Prev</a>
                </div>
            
        </div>
        <div class="next-item">
            
                <div class="icon arrow-right"></div>
                <div class="post-link">
                  <a href="/2025/04/20/linux-amd-screen-boom/">Next</a>  
                </div>  
            
        </div>
    </nav>
</div>

    
      <div class="post-comment">

     

     
    
    

</div>
     
  
</article>
        </div>
      </div>
      
      <div class="footer">
    <div class="flex-container">
        <div class="footer-text">
            
                        
                            韩暮秋 |
                                
                                    
                                        希望路过的人可以添点柴火让这里暖和点
                                        
        </div>
    </div>
</div>


    </div>
    
    <script src="/js/mermaid-zoom.js"></script>

  </body>
</html>