基本上，每个客户都有一个 clientId 与它们相关联的消息---可以推送消息，重要的是，在第一条消息完成处理之前，不会处理来自同一客户端的第二条消息（即使客户端可以连续发送多条消息，并且它们是有序的，并且发送消息的多个客户端最好不要相互干扰）。而且，重要的是，一份工作不应该被处理两次。
我认为使用redis可能可以解决这个问题，我开始使用bull库进行一些快速原型设计，但我显然做得不好，我希望有人知道如何继续。

这就是我目前所尝试的：

使用 clientId 作为作业名称。
在两个独立进程上等待大量随机时间时消耗作业。
我尝试添加我正在使用的库提供的默认锁定( bull )但是它锁定jobid，jobid对于每个作业都是唯一的，而不是clientid。

我希望发生什么：

其中一个消费者不能接受同样的工作 clientId 直到上一个完成处理。
但是，他们应该能够从不同的服务器获取项目 clientId 在没有问题的情况下（异步地）并行(我还没走到这一步，我现在只处理一个 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 &