@efebia/队列
用于构建消息队列消费者的抽象基础类,并有内置的内置执行.
安装
npm install @efebia/queue
使用量
扩展 Queue
执行三种抽象方法连接任何队列后端:
import { Queue, QueueCallback, SentMessage } from '@efebia/queue';
type MyMessage = { userId: string; action: string };
class MyQueue extends Queue<RawSQSMessage, MyMessage> {
async readMessages() {
// fetch messages from your backend, return [] or undefined if none
return await fetchFromBackend();
}
async processMessage(raw: RawSQSMessage, callback?: QueueCallback<MyMessage>) {
const parsed = parse(raw);
const state = { del: true };
await callback?.(parsed.name, parsed.params, state);
if (state.del) await deleteFromBackend(raw);
}
async sendMessage(body: SentMessage<MyMessage>) {
await sendToBackend(JSON.stringify(body));
}
}
const queue = new MyQueue({ sleepTimeout: 3000 });
// Start the polling loop
queue.start(async (name, params, state) => {
console.log(name, params);
// set state.del = false to prevent deletion
});
使用 MemoryQueue
用于测试或简单使用例的即用内置队列:
import { MemoryQueue } from '@efebia/queue';
type Job = { task: string };
const queue = new MemoryQueue<Job>({});
// Send a message
await queue.send({ name: 'doWork', params: { task: 'hello' } });
// Start consuming
queue.start(async (name, params, state) => {
console.log(`Processing ${name}:`, params);
});
// Stop the loop
queue.stop();
API 密码
Queue<TReceivedMessage, TMessage> (摘要)
构造选项( E)QueueOptions:
| 选项 | 类型 | 默认 | 说明 |
|---|---|---|---|
sleepTimeout | number | 5000 | 重试前等待出错的毫秒 |
errorCallback | (error) => void | console.error | 在投票循环中出现未发现的错误 |
执行摘要方法:
| 方法 | 说明 |
|---|---|
readMessages() | 获取下一批信件 。 返回时 undefined 以跳过当前迭代. |
processMessage(message, callback?) | 处理单个消息并引用回调 . |
sendMessage(body) | 向后面的队列后端发送信息 . |
方法:
start(callback, canThrow?)开始无限的投票循环stop()—— 指示当前迭代后要退出的循环resume(callback)——重新启动已停止的队列isRunning()——返回true如果循环活动send(message)——发送信息,自动添加createdAt改为参数
回调签名 :
(name: string, params: TMessage, state: { del: boolean }) => void | Promise<void>
设定 state.del = false 在回调中防止处理后信件被删除.
MemoryQueue<TMessage>
扩展 Queue 带有内存信件阵列。 可用于测试和地方发展.
类型
Message<TParams>—— 说{ name: string; params: TParams }SentMessage<TParams>—— 说Message与params.createdAt: Date已添加
许可证
麻省理工学院