基本上，每个客户端（clientId与它们关联）都可以推送消息，重要的是，在第一个消息完成处理之前，不会处理来自同一客户端的第二条消息（即使客户端可以连续发送多条消息，并且它们是有序的，并且多个发送消息的客户端理想情况下不应相互干扰）。而且，重要的是，一个工作不应该被处理两次。
我认为使用 Redis 可能能够解决这个问题，我开始使用 Bull 库进行一些快速原型设计，但我显然做得不好，我希望有人知道如何继续。
这是我到目前为止所尝试的：
- 创建作业并将它们添加到一个进程的相同队列名称中，使用
clientId作为作业名称。 - 在 2 个单独的进程上等待大量随机时间的同时消耗作业。
- 我尝试添加我正在使用的库提供的默认锁定 (
bull)，但它锁定在 jobId 上，它对每个作业都是唯一的，而不是在 clientId 上。
我想要发生的事情：
clientId在前一个消费者完成处理之前，其中一个消费者不能从同一个消费者那里接手工作。- 但是，它们应该能够
clientId毫无问题地（异步地）并行地从不同的 s 中获取项目。（我还没有走到这一步，我现在只处理一个clientId）
我得到什么：
- 两个消费者都从队列中消费尽可能多的项目，而无需等待前一个项目
clientId完成。
Redis 甚至是适合这项工作的工具吗？
示例代码
// ./setup.ts
import Queue from 'bull';
import * as uuid from 'uuid';
// Check that when a message is taken from a place, no other message is taken
// TO do that test, have two processes that process messages and one that sets messages, and make the job take a long time
// queue for each room https://stackoverflow.com/questions/54178462/how-does-redis-pubsub-subscribe-mechanism-works/54243792#54243792
// https://groups.google.com/forum/#!topic/redis-db/R09u__3Jzfk
// Make a job not be called stalled, waiting enough time https://github.com/OptimalBits/bull/issues/210#issuecomment-190818353
export async function sleep(ms: number): Promise<void> {
return new Promise((resolve) => {
setTimeout(resolve, ms);
});
}
export interface JobData {
id: string;
v: number;
}
export const queue = new Queue<JobData>('messages', 'redis://127.0.0.1:6379');
queue.on('error', (err) => {
console.error('Uncaught error on queue.', err);
process.exit(1);
});
export function clientId(): string {
return uuid.v4();
}
export function randomWait(minms: number, maxms: number): Promise<void> {
const ms = Math.random() * (maxms - minms) + minms;
return sleep(ms);
}
// Make a job not be called stalled, waiting enough time https://github.com/OptimalBits/bull/issues/210#issuecomment-190818353
// eslint-disable-next-line @typescript-eslint/ban-ts-comment
//@ts-ignore
queue.LOCK_RENEW_TIME = 5 * 60 * 1000;
// ./create.ts
import { queue, randomWait } from './setup';
const MIN_WAIT = 300;
const MAX_WAIT = 1500;
async function createJobs(n = 10): Promise<void> {
await randomWait(MIN_WAIT, MAX_WAIT);
// always same Id
const clientId = Math.random() > 1 ? 'zero' : 'one';
for (let index = 0; index < n; index++) {
await randomWait(MIN_WAIT, MAX_WAIT);
const job = { id: clientId, v: index };
await queue.add(clientId, job).catch(console.error);
console.log('Added job', job);
}
}
export async function create(nIds = 10, nItems = 10): Promise<void> {
const jobs = [];
await randomWait(MIN_WAIT, MAX_WAIT);
for (let index = 0; index < nIds; index++) {
await randomWait(MIN_WAIT, MAX_WAIT);
jobs.push(createJobs(nItems));
await randomWait(MIN_WAIT, MAX_WAIT);
}
await randomWait(MIN_WAIT, MAX_WAIT);
await Promise.all(jobs)
process.exit();
}
(function mainCreate(): void {
create().catch((err) => {
console.error(err);
process.exit(1);
});
})();
// ./consume.ts
import { queue, randomWait, clientId } from './setup';
function startProcessor(minWait = 5000, maxWait = 10000): void {
queue
.process('*', 100, async (job) => {
console.log('LOCKING: ', job.lockKey());
await job.takeLock();
const name = job.name;
const processingId = clientId().split('-', 1)[0];
try {
console.log('START: ', processingId, '\tjobName:', name);
await randomWait(minWait, maxWait);
const data = job.data;
console.log('PROCESSING: ', processingId, '\tjobName:', name, '\tdata:', data);
await randomWait(minWait, maxWait);
console.log('PROCESSED: ', processingId, '\tjobName:', name, '\tdata:', data);
await randomWait(minWait, maxWait);
console.log('FINISHED: ', processingId, '\tjobName:', name, '\tdata:', data);
} catch (err) {
console.error(err);
} finally {
await job.releaseLock();
}
})
.catch(console.error); // Catches initialization
}
startProcessor();
这是使用 3 个不同的进程运行的，您可能会这样称呼它们（尽管我使用不同的选项卡来更清楚地了解正在发生的事情）
npx ts-node consume.ts &
npx ts-node consume.ts &
npx ts-node create.ts &