队列

node ace add @adonisjs/queue

import env from '#start/env'
import { defineConfig, drivers } from '@adonisjs/queue'

export default defineConfig({
  default: env.get('QUEUE_DRIVER', 'redis'),

  adapters: {
    redis: drivers.redis({
      connectionName: 'main',
    }),
    sync: drivers.sync(),
  },

  worker: {
    concurrency: 5,
    idleDelay: '2s',
  },

  locations: ['./app/jobs/**/*.{ts,js}'],
})

import { defineConfig, drivers } from '@adonisjs/queue'

export default defineConfig({
  default: 'redis',
  adapters: {
    redis: drivers.redis({
      // 使用 config/redis.ts 中的 'main' 连接
      connectionName: 'main',
    }),
  },
  // ...
})

import { defineConfig, drivers } from '@adonisjs/queue'

export default defineConfig({
  default: 'database',
  adapters: {
    database: drivers.database({
      connectionName: 'primary',
    }),
  },
  // ...
})

import { BaseSchema } from '@adonisjs/lucid/schema'
import { QueueSchemaService } from '@adonisjs/queue'

export default class extends BaseSchema {
  async up() {
    const schemaService = new QueueSchemaService(this.db.getWriteClient())

    await schemaService.createJobsTable()
    await schemaService.createSchedulesTable()
  }

  async down() {
    const schemaService = new QueueSchemaService(this.db.getWriteClient())

    await schemaService.dropSchedulesTable()
    await schemaService.dropJobsTable()
  }
}

import { defineConfig, drivers } from '@adonisjs/queue'

export default defineConfig({
  default: 'sync',
  adapters: {
    sync: drivers.sync(),
  },
  // ...
})

node ace make:job process_payment

import { Job } from '@adonisjs/queue'
import type { JobOptions } from '@adonisjs/queue/types'

interface ProcessPaymentPayload {
  orderId: number
  amount: number
  currency: string
}

export default class ProcessPayment extends Job<ProcessPaymentPayload> {
  static options: JobOptions = {
    queue: 'default',
    maxRetries: 3,
  }

  async execute() {
    const { orderId, amount, currency } = this.payload

    // 使用你的支付网关处理支付
    console.log(`Processing payment of ${amount} ${currency} for order ${orderId}`)
  }

  async failed(error: Error) {
    console.error(`Payment processing failed for order ${this.payload.orderId}:`, error.message)
  }
}

export default class ProcessPayment extends Job<ProcessPaymentPayload> {
  async execute() {
    // 访问任务元数据
    console.log(`Job ID: ${this.context.jobId}`)
    console.log(`Attempt: ${this.context.attempt}`)
    console.log(`Queue: ${this.context.queue}`)
    console.log(`Priority: ${this.context.priority}`)
  }
}

import { Job } from '@adonisjs/queue'
import type { JobOptions } from '@adonisjs/queue/types'

interface GenerateReportPayload {
  reportId: number
  rows: number[]
}

export default class GenerateReport extends Job<GenerateReportPayload> {
  static options: JobOptions = {
    timeout: '5m',
    failOnTimeout: false, // 重试而非永久失败
  }

  async execute() {
    for (const rowId of this.payload.rows) {
      // 在处理每一行之前检查任务是否已超时
      if (this.signal?.aborted) {
        throw new Error('Job timed out during report generation')
      }

      await this.processRow(rowId)
    }
  }

  private async processRow(rowId: number) {
    // 处理单个行
  }
}

import { inject } from '@adonisjs/core'
import { Job } from '@adonisjs/queue'
import type { JobOptions } from '@adonisjs/queue/types'
import PaymentService from '#services/payment_service'

interface ProcessPaymentPayload {
  orderId: number
  amount: number
  currency: string
}

@inject()
export default class ProcessPayment extends Job<ProcessPaymentPayload> {
  static options: JobOptions = {
    queue: 'payments',
    maxRetries: 3,
  }

  constructor(private paymentService: PaymentService) {
    super()
  }

  async execute() {
    await this.paymentService.charge(
      this.payload.orderId,
      this.payload.amount,
      this.payload.currency
    )
  }
}

import ProcessPayment from '#jobs/process_payment'

export default class OrdersController {
  async store({ request, response }: HttpContext) {
    const order = await Order.create(request.body())

    // 派发任务以进行后台处理
    await ProcessPayment.dispatch({
      orderId: order.id,
      amount: order.total,
      currency: 'USD',
    })

    return response.created(order)
  }
}

import { BaseCommand } from '@adonisjs/core/ace'
import ImportOrders from '#jobs/import_orders'

export default class ImportOrdersCommand extends BaseCommand {
  static commandName = 'orders:import'

  static options = {
    startApp: true,
  }

  async run() {
    const queue = await this.app.container.make('queue.manager')

    await queue.loadJobs()

    await ImportOrders.dispatch({
      source: 'warehouse',
    })
  }
}

// 派发到特定队列并设为高优先级
await ProcessPayment.dispatch({
  orderId: order.id,
  amount: order.total,
  currency: 'USD',
})
  .toQueue('payments')
  .priority(1)

import type { HttpContext } from '@adonisjs/core/http'
import Order from '#models/order'
import ProcessPayment from '#jobs/process_payment'

export default class OrdersController {
  async retryPayment({ params, response }: HttpContext) {
    const order = await Order.findOrFail(params.id)

    const result = await ProcessPayment.dispatch({
      orderId: order.id,
      amount: order.total,
      currency: order.currency,
    })
      .toQueue('payments')
      .dedup({ id: `order:${order.id}`, ttl: '10m' })

    return response.accepted({
      jobId: result.jobId,
      deduped: result.deduped,
    })
  }
}

import SendNewsletter from '#jobs/send_newsletter'

export default class NewsletterService {
  async sendToAllSubscribers(subscribers: { email: string }[]) {
    const payloads = subscribers.map((subscriber) => ({
      email: subscriber.email,
      subject: 'Monthly Newsletter',
    }))

    const { jobIds } = await SendNewsletter.dispatchMany(payloads)
      .toQueue('emails')
      .group('newsletter-march-2025')

    console.log(`Dispatched ${jobIds.length} newsletter jobs`)
  }
}

import { defineConfig, drivers } from '@adonisjs/queue'
import { exponentialBackoff } from '@adonisjs/queue'

export default defineConfig({
  default: 'redis',
  adapters: {
    redis: drivers.redis({ connectionName: 'main' }),
  },
  retry: {
    maxRetries: 3,
    backoff: exponentialBackoff(),
  },
  // ...
})

import { exponentialBackoff } from '@adonisjs/queue'

// 默认：1s 基准、5m 最大、2x 倍数、启用抖动
exponentialBackoff()

// 自定义
exponentialBackoff({ baseDelay: '500ms', maxDelay: '1m' })

import { linearBackoff } from '@adonisjs/queue'

// 默认：5s 基准、2m 最大
linearBackoff()

// 自定义
linearBackoff({ baseDelay: '10s', maxDelay: '5m' })

import { fixedBackoff } from '@adonisjs/queue'

// 默认：10s
fixedBackoff()

// 自定义
fixedBackoff('30s')

import { customBackoff } from '@adonisjs/queue'

customBackoff({
  strategy: 'exponential',
  baseDelay: '100ms',
  maxDelay: '30s',
  multiplier: 3,
  jitter: false,
})

import { Job } from '@adonisjs/queue'
import { exponentialBackoff } from '@adonisjs/queue'
import type { JobOptions } from '@adonisjs/queue/types'

export default class ProcessPayment extends Job<ProcessPaymentPayload> {
  static options: JobOptions = {
    maxRetries: 5,
    retry: {
      backoff: exponentialBackoff({ baseDelay: '2s', maxDelay: '10m' }),
    },
  }

  async execute() {
    // 支付处理逻辑
  }
}

import CleanupExpiredSessions from '#jobs/cleanup_expired_sessions'
import GenerateWeeklyReport from '#jobs/generate_weekly_report'

// 每天午夜运行
await CleanupExpiredSessions.schedule({ retentionDays: 30 })
  .cron('0 0 * * *')
  .timezone('Europe/Paris')

// 每周一上午 9 点运行
await GenerateWeeklyReport.schedule({ type: 'weekly' })
  .cron('0 9 * * MON')
  .timezone('America/New_York')

import SyncInventory from '#jobs/sync_inventory'

// 每 5 分钟运行一次
await SyncInventory.schedule({ source: 'warehouse-api' })
  .every('5m')

import { Schedule } from '@adonisjs/queue'

export default class AdminController {
  async listSchedules() {
    // 列出所有计划
    const schedules = await Schedule.list()

    // 仅列出活跃的计划
    const active = await Schedule.list({ status: 'active' })

    return schedules
  }

  async pauseSchedule({ params }: HttpContext) {
    const schedule = await Schedule.find(params.id)

    if (schedule) {
      await schedule.pause()
    }
  }

  async resumeSchedule({ params }: HttpContext) {
    const schedule = await Schedule.find(params.id)

    if (schedule) {
      await schedule.resume()
    }
  }

  async triggerSchedule({ params }: HttpContext) {
    const schedule = await Schedule.find(params.id)

    if (schedule) {
      // 立即派发定时任务
      await schedule.trigger()
    }
  }

  async deleteSchedule({ params }: HttpContext) {
    const schedule = await Schedule.find(params.id)

    if (schedule) {
      await schedule.delete()
    }
  }
}

# 列出所有定时任务
node ace queue:scheduler:list

# 仅列出活跃的计划
node ace queue:scheduler:list --status=active

# 移除特定的计划
node ace queue:scheduler:remove <schedule-id>

# 移除所有计划
node ace queue:scheduler:clear

node ace queue:work

# 处理特定的队列
node ace queue:work --queue=payments,emails

# 设置并发数（同时处理的任务数量）
node ace queue:work --concurrency=10

export default defineConfig({
  // ...
  worker: {
    concurrency: 5,
    idleDelay: '2s',
    stalledThreshold: '30s',
    stalledInterval: '30s',
    maxStalledCount: 1,
    gracefulShutdown: true,
  },
})

import { test } from '@japa/runner'
import { QueueManager } from '@adonisjs/queue'
import ProcessPayment from '#jobs/process_payment'

test.group('Orders', (group) => {
  group.each.teardown(() => {
    QueueManager.restore()
  })

  test('dispatches a payment job when creating an order', async ({ client }) => {
    const fake = QueueManager.fake()

    const response = await client.post('/orders').json({
      product_id: 1,
      quantity: 2,
    })

    response.assertStatus(201)

    // 断言任务已被派发
    fake.assertPushed(ProcessPayment)

    // 使用负载匹配进行断言
    fake.assertPushed(ProcessPayment, {
      payload: { orderId: 1, amount: 100, currency: 'USD' },
    })
  })

  test('does not dispatch a payment job for free orders', async ({ client }) => {
    const fake = QueueManager.fake()

    await client.post('/orders').json({
      product_id: 1,
      quantity: 0,
    })

    fake.assertNotPushed(ProcessPayment)
  })
})

import { test } from '@japa/runner'
import { QueueManager } from '@adonisjs/queue'
import SendNewsletter from '#jobs/send_newsletter'

test('dispatches newsletter jobs for all subscribers', async ({ client }) => {
  const fake = QueueManager.fake()

  await client.post('/newsletters/send')

  // 使用函数进行匹配
  fake.assertPushed(SendNewsletter, {
    payload: (payload) => payload.email.endsWith('@example.com'),
  })

  // 检查延迟
  fake.assertPushed(SendNewsletter, {
    delay: (delay) => delay !== undefined && delay > 0,
  })
})