@arkstack/队列

Arkstack的队列模块,提供基于驱动的队列运输和工人进行背景处理.

@arkstack/queue 这是 运输 层:连接、司机和工人。 撰写您的工作 @arkstack/jobs,提供 Job 基类和 a dispatch() 此软件包上的助手 .

连接

配置

// src/config/queue.ts
import type { QueueConfig } from '@arkstack/queue'

export default (): QueueConfig => ({
  default: env('QUEUE_CONNECTION', 'sync'),
  connections: {
    sync: { driver: 'sync' },
    database: { driver: 'database', table: 'jobs', queue: 'default' },
    redis: {
      driver: 'redis',
      host: env('REDIS_HOST', '127.0.0.1'),
      port: env('REDIS_PORT', 6379),
    },
  },
})

使用量

import { Queue } from '@arkstack/queue'

await Queue.push(new SendWelcomeEmail(user)) // onto the default connection
await Queue.later(60, new ChargeInvoice(invoiceId)) // after 60 seconds
await Queue.connection('redis').push(job, 'emails') // a specific connection/queue
await Queue.size('emails')
await Queue.clear('emails')

工人

ark queue:work                      # daemon on the default connection's default queue
ark queue:work redis --queue=emails
ark queue:work --queue=high,default # priority list: high first, then default
ark queue:work '--queue=*'          # every queue on the connection (quote the *)
ark queue:work --once               # process a single job and exit
ark queue:clear redis --queue=emails

--queue 取一个队列,一个逗号分隔列表,或 *数字 :

• 没有 --queue: 只有连接的默认队列( t)queue . . . . . . . 工作 发往别处 onQueue(...) 或一个工作 queue 属性为 没有 已处理.
• 列表 (单位:千美元)--queue=webhook,ocr,default)是严格的优先级顺序: 环绕工人的第一个工作,它发现在行走列表,所以以后排队 运行时,前一个是空的.
• * 运行连接上的每个队列。 这个 database 驱动程序 所有队列中最老的应交工作 redis 驱动程序发现队列 SCAN 和工作他们圆 robin。 将它与名字合并,以优先排序一些: --queue=webhook,*。引用任意 * 在外壳中('--queue=webhook,*'),或 zsh/bash可能将其扩展为被子.

每一个获得工作的队列都需要一个列出工作的工人(或 *),或其工作 永远等待.

在方案方面:

import { Queue } from '@arkstack/queue'

const worker = Queue.worker('database')
await worker.daemon({ queue: 'default', sleep: 3 })
await worker.daemon({ queue: ['webhook', '*'], sleep: 3 }) // priority list / wildcard

自定义驱动程序支持 * 以压倒性 popAny() 打开 QueueContract; 基础 执行只执行默认队列.

一个工作是 删除 关于成功, 已发行 对于又一次尝试失败,和 标记失败 (援引国际法院 failed 钩子) 一旦它耗尽 tries。 。 。 .

任务 (去) 序列化

司机以外的司机 sync 存储序列载荷并需要重建任务 工人身上的情况. @arkstack/jobs 为您注册这些策略 手动操作:

import { Queue } from '@arkstack/queue'

Queue.serializeUsing((job) => ({/* JobPayload */}))
Queue.resolveJobsUsing((payload) => rebuildJob(payload))