References: Gregor Hohpe, Enterprise Integration Patterns; Stripe Workflows practices; Uber Cadence/Temporal; Outbox Pattern (Chris Richardson).
Applicable Scenarios
- Time-consuming operations (sending emails, generating reports, video transcoding)
- Third-party API integration (async retries)
- Decoupling services (event-driven)
- Peak shaving
- Scheduled tasks (cron)
- Workflow orchestration (multi-step business processes)
Core Principles
1. 任务必须幂等
At-least-once 投递 = 至少一次 = 可能多次
-
Tasks must be retryable Have a max attempts + retry interval policy
-
Tasks must have a dead letter queue After max retries, go to dead letter for manual intervention
-
Task parameters small (< 256KB) Use ID references for large objects, don't pass complete data
-
Don't send messages inside a transaction Use the Outbox pattern to ensure DB + MQ consistency
-
Distinguish real-time requirements Latency < 1s: in-memory queue Latency < 10s: Redis Streams Latency acceptable: RabbitMQ / Kafka
-
Monitor core metrics Queue depth / processing latency / failure rate / dead letter count
## 消息队列选型
| 方案 | 适合场景 | 吞吐 | 持久化 | 事务 |
|---|---|---|---|---|
| **RabbitMQ** | 业务任务、复杂路由 | 万级 | ✅ | ✅ |
| **Apache Kafka** | 事件流、大数据 | 百万级 | ✅ | ✅ |
| **Redis Streams** | 轻量、Redis 一站式 | 十万级 | ✅ | 弱 |
| **AWS SQS** | 托管、Serverless | 万级 | ✅ | - |
| **NATS** | 微服务间通信 | 百万级 | 可选 | - |
| **Apache Pulsar** | 多租户、Kafka 替代 | 百万级 | ✅ | ✅ |
## 任务队列框架
| 框架 | 语言 | 后端 | 特点 |
|---|---|---|---|
| **Bull / BullMQ** | TS | Redis | NestJS 友好、UI 好 |
| **Celery** | Python | Redis / RabbitMQ | Python 标杆、生态全 |
| **Sidekiq** | Ruby | Redis | Rails 标配 |
| **Spring Batch** | Java | DB-based | 企业级批处理 |
| **Spring @Async** | Java | 内存（默认）/ MQ | 轻量异步 |
| **Asynq** | Go | Redis | Go 任务队列 |
| **Hangfire** | C# | DB-based | .NET 标配 |
| **Temporal** | 多语言 | 自建 | 工作流引擎、复杂编排 |
## NestJS + BullMQ 范式
```typescript
// 1. 配置队列
@Module({
imports: [
BullModule.registerQueue({
name: 'email',
defaultJobOptions: {
attempts: 3,
backoff: { type: 'exponential', delay: 1000 },
removeOnComplete: 100,
removeOnFail: 1000,
},
}),
],
})
export class AppModule {}
// 2. 生产者
@Injectable()
export class OrderService {
constructor(@InjectQueue('email') private emailQueue: Queue) {}
async createOrder(dto: CreateOrderDto) {
const order = await this.repo.create(dto);
// 发送任务（不阻塞主流程）
await this.emailQueue.add('send-confirmation', {
orderId: order.id,
userEmail: order.user.email,
}, {
jobId: `email:${order.id}`, // 幂等键
delay: 5000, // 5 秒后执行
});
return order;
}
}
// 3. 消费者
@Processor('email')
export class EmailProcessor {
constructor(private mailService: MailService) {}
@Process('send-confirmation')
async handleSendConfirmation(job: Job<{orderId: number; userEmail: string}>) {
const { orderId, userEmail } = job.data;
// 幂等检查
const sent = await this.checkSent(orderId);
if (sent) {
return { skipped: true };
}
await this.mailService.sendConfirmation(userEmail, orderId);
await this.markSent(orderId);
}
@OnQueueFailed()
onFailed(job: Job, err: Error) {
if (job.attemptsMade >= job.opts.attempts) {
// 进入死信，告警
this.alertService.notify('Email job failed after retries', { jobId: job.id, err });
}
}
}
Spring + RabbitMQ Paradigm
// 1. 配置
@Configuration
public class RabbitConfig {
@Bean
public Queue emailQueue() {
return QueueBuilder.durable("email.queue")
.withArgument("x-dead-letter-exchange", "dlx")
.withArgument("x-dead-letter-routing-key", "email.dead")
.withArgument("x-message-ttl", 3600000)
.build();
}
@Bean
public Queue emailDeadLetterQueue() {
return QueueBuilder.durable("email.dead").build();
}
}
// 2. 生产者
@Service
public class OrderService {
@Autowired
private RabbitTemplate rabbitTemplate;
@Transactional
public Order createOrder(CreateOrderRequest req) {
Order order = orderRepo.save(...);
// 用 Outbox 模式（详见下文）
outboxRepo.save(new OutboxEvent("email.send", order.getId()));
return order;
}
}
// 3. 消费者
@Component
public class EmailListener {
@RabbitListener(queues = "email.queue")
public void handleEmail(EmailMessage message) {
try {
// 幂等检查
if (isAlreadySent(message.getJobId())) return;
mailService.send(message);
markSent(message.getJobId());
} catch (RetryableException e) {
throw new AmqpRejectAndDontRequeueException(e); // 进死信
}
}
@RabbitListener(queues = "email.dead")
public void handleDeadLetter(EmailMessage message) {
log.error("Dead letter: {}", message);
alertService.notify("Email DLQ", message);
}
}
Python + Celery Paradigm
# 1. 配置
from celery import Celery
app = Celery('myapp', broker='redis://localhost:6379/0', backend='redis://localhost:6379/1')
app.conf.update(
task_acks_late=True, # 任务完成后才 ack（防丢）
task_reject_on_worker_lost=True,
worker_prefetch_multiplier=1, # 公平分发
task_default_retry_delay=60,
task_max_retries=3,
)
# 2. 任务定义
@app.task(
bind=True,
autoretry_for=(RetryableError,),
retry_backoff=True,
retry_backoff_max=300,
max_retries=5,
)
def send_email(self, order_id: int, user_email: str):
if is_already_sent(order_id):
return {'skipped': True}
try:
mail_service.send(user_email, order_id)
mark_sent(order_id)
except SmtpTransientError as e:
# 重试
raise self.retry(exc=e)
except SmtpPermanentError as e:
# 不重试，进死信
log.error(f'Permanent error: {e}')
raise
# 3. 生产
@transaction.atomic
def create_order(...):
order = Order.objects.create(...)
# 用 transaction.on_commit 保证 DB 提交后才发任务
transaction.on_commit(
lambda: send_email.apply_async(
args=[order.id, order.user.email],
countdown=5, # 5 秒后
)
)
return order
Go + Asynq Paradigm
// 任务定义
type EmailTask struct {
OrderID int64
Email string
}
// 生产
func (s *OrderService) CreateOrder(ctx context.Context, req CreateOrderRequest) error {
// 在事务里
return s.db.Transaction(func(tx *gorm.DB) error {
order := &Order{...}
if err := tx.Create(order).Error; err != nil {
return err
}
// Outbox：写到事件表
return tx.Create(&Outbox{
Type: "email.send",
Payload: mustJson(EmailTask{OrderID: order.ID, Email: order.Email}),
}).Error
})
}
// 消费（独立进程）
func handleEmail(ctx context.Context, t *asynq.Task) error {
var task EmailTask
if err := json.Unmarshal(t.Payload(), &task); err != nil {
return fmt.Errorf("unmarshal: %w", err)
}
// 幂等
if alreadySent(task.OrderID) {
return nil
}
if err := mailService.Send(task.Email, task.OrderID); err != nil {
return fmt.Errorf("send: %w", err) // 自动重试
}
return markSent(task.OrderID)
}
// Worker
srv := asynq.NewServer(redisOpt, asynq.Config{
Concurrency: 10,
Queues: map[string]int{"critical": 6, "default": 3, "low": 1},
})
mux := asynq.NewServeMux()
mux.HandleFunc("email.send", handleEmail)
srv.Run(mux)
Outbox Pattern (solving transactional consistency)
问题：
在事务里 send 消息：
BEGIN;
INSERT order;
SEND email message; ← 如果发送失败但 commit 了，怎么办？
COMMIT;
消息发送和事务无法原子化。
Outbox 解决：
1. 业务操作 + 写入 outbox 表（同一事务）
BEGIN;
INSERT order;
INSERT outbox(type='email', payload=...);
COMMIT;
2. 独立进程定期扫 outbox 表，发送消息
SELECT * FROM outbox WHERE published_at IS NULL;
SEND message;
UPDATE outbox SET published_at = NOW() WHERE id = ?;
3. 也可用 CDC（Debezium 监听 binlog）
优势：
- 业务和消息原子一致
- 消息发送失败可重试
- 可审计
CREATE TABLE outbox (
id bigserial PRIMARY KEY,
event_type varchar(64) NOT NULL,
payload jsonb NOT NULL,
created_at timestamptz DEFAULT now(),
published_at timestamptz,
attempts integer DEFAULT 0
);
CREATE INDEX idx_outbox_unpublished ON outbox(created_at)
WHERE published_at IS NULL;
Workflow Orchestration (Temporal / Cadence)
For complex multi-step processes (e.g., order → payment → inventory → logistics), use Temporal:
// Workflow 定义
export async function orderFulfillment(orderId: number): Promise<void> {
await wf.workflowInfo();
try {
await wf.executeActivity('reserveInventory', { orderId, scheduleToCloseTimeout: '1m' });
await wf.executeActivity('processPayment', { orderId, scheduleToCloseTimeout: '5m' });
await wf.executeActivity('shipOrder', { orderId, scheduleToCloseTimeout: '1d' });
} catch (err) {
// 自动补偿
await wf.executeActivity('refund', { orderId });
await wf.executeActivity('releaseInventory', { orderId });
throw err;
}
}
Retry Policy
立即重试：1 次
指数退避：
attempt 1: wait 1s
attempt 2: wait 4s
attempt 3: wait 16s
attempt 4: wait 64s
attempt 5: 进死信
抖动（防雷鸣）：
wait = base * 2^attempt + random(0, 1000ms)
// BullMQ 配置
{
attempts: 5,
backoff: { type: 'exponential', delay: 1000 }, // 1s, 2s, 4s, 8s, 16s
}
Scheduled Tasks
// NestJS Schedule
@Injectable()
export class CronService {
@Cron('0 0 * * *') // 每天 0 点
async dailyReport() {
await this.reportService.generate();
}
@Interval(60000) // 每分钟
async healthCheck() {
await this.checkUpstream();
}
}
# Celery Beat
app.conf.beat_schedule = {
'daily-report': {
'task': 'tasks.daily_report',
'schedule': crontab(hour=0, minute=0),
},
}
Workflow
1. 评估是否需要异步
- 主流程实时性？
- 失败可接受性？
- 用户感知？
2. 选择队列方案
- 简单 → BullMQ / Asynq / Sidekiq
- 复杂 → RabbitMQ / Kafka
- 工作流 → Temporal
3. 设计任务
- 参数小（用 ID 引用）
- 幂等（用业务唯一键）
- 可重试
4. 设计死信处理
- 监控告警
- 人工介入流程
5. 选择事务一致性方案
- 简单：transaction.on_commit
- 严格：Outbox 模式
6. 监控
- 队列深度
- 处理延迟 P99
- 失败率
- 死信积压
7. 测试
- 重试场景
- 并发场景
- 死信场景
Supporting Templates
templates/async-job-template.md— Task design + idempotency + retry + dead letter + Outbox + monitoring
Quality Self-Check
□ 任务幂等（同任务多次执行结果一致）
□ 重试策略明确（次数 + 间隔）
□ 死信队列存在 + 告警
□ 任务参数 < 256KB
□ 大数据用 ID 引用
□ 不在事务里发消息（用 Outbox 或 on_commit）
□ 监控：队列深度 / 延迟 / 失败率
□ 优雅关闭（Worker 处理完再退出）
□ 任务超时设置
□ 队列分级（critical / default / low）
Common Pitfalls
- Tasks not idempotent——duplicate execution charges twice
- Not using a dead letter queue——failed tasks are lost
- Task parameters too large——queue explosion
- Sending messages inside a transaction——DB rolls back but message already sent
- Not monitoring queue depth——only discover the backlog when it blows up
- Unlimited retries——zombie tasks
- Not distinguishing retryable vs non-retryable errors——SMTP rejection keeps retrying
- Worker not gracefully shutting down——task killed halfway
- No concurrency limit——OOM
- Mixing long and short tasks in the same queue——long tasks block
- Scheduled task executed repeatedly by multiple instances——add a distributed lock
- Tasks have no timeout——hang forever without releasing
- Mock tasks not sent to the test environment——only discovered in production
Collaboration with Other Skills
上游：
api-implementation → 接口提交任务
data-access → Outbox 写库
下游：
error-handling-resilience → 重试 / 熔断
observability → 任务监控
testing-implementation → Mock 队列
Related References
- Project root
references/backend-tech-stack-guide.md— Queue selection