One job at a time in workers?

[SupertigerDev]

index.mjs:

import {Worker} from 'node:worker_threads'
import {cpus} from 'os'

cpus().forEach(() => {  // 2 cpus
    const worker = new Worker('./worker.mjs');
})

worker.mjs

import Queue from 'bull';
import {setTimeout} from 'timers/promises'

const queue = new Queue('queue', "redis://127.0.0.1:6379");


queue.process(async (job) => {
    console.log("start")
    await setTimeout(5000);
    console.log("end");
    return {done: "Done!"}
})

queue.add({})

This responds like this:

start // worker 1
start // worker 2
end // worker 1
end //worker 2

They dont run one at a time. How could I fix this? It should return like this:

start // worker 1
end // worker 1
start // worker 2
end  // worker 2

I currently serve an express server in multiple workers to load balance my server. I would like to also add this queuing system but be able to run the job in the worker where it was started from.

[roggervalf]

hi @SupertigerDev, what you are experiencing is expected as you are creating 2 workers, each worker has a concurrency value 1 by default. Each of them will pick 1 job to be processed. Looks like what you are looking for is a global concurrency, I highly recommend you to use bullmq package https://docs.bullmq.io/guide/queues/global-concurrency where this feature is available

[roggervalf]

not sure what you meant by downgrating to v2 🤔 , but global concurrency is a feature that was recently added. If you need something like concurrency per id, this sounds like using our groups concept that is better implemented in bullmq-pro https://docs.bullmq.io/bullmq-pro/groups/concurrency

[roggervalf]

Ok I see what you meant, yes old groups logic was removed because it didn't allow is to handle a global rate limit in an efficient way and also because it was reimplemented in order to be able to add more features to group logics

[manast]

@SupertigerDev BullMQ Pro supports groups and I think the features that you need: https://docs.bullmq.io/bullmq-pro/groups

[SupertigerDev]

Seems like it costs money 💀

I just made my own library: https://github.com/Nerimity/Mimiqueue/

Mine works like this:

import {AltQueue} from '@nerimity/mimiqueue';

const someQueue = new AltQueue({redisClient, name: 'queue'});

async function doSomething(userId: string) {
   const finish = await someQueue.start({groupName: userId});
   
   // do something...
   
   finish();
}
doSomething("123");
doSomething("123");

It seems to be working how I want 😄

[manast]

Build or buy! 😅