feat: concurrency control queue

[kanzihuang]

No description provided.

[kamikazechaser]

@kanzihuang I assume this is still a WIP because I don't see the queue concurrency value being used in the processor/semaphore. I assume that is the end goal?

[gsaraf]

@kamikazechaser - Hi! Reviewing the code, I think it does work as intended - instead of modifying the way the processor or semaphore work (which would affect only that specific Server), it uses the number of currently-being-processed tasks by checking its cardinality compared to the maximum concurrency for that queue.

If you are looking for a different solution I'd be happy to take a look at it - I'd love for the dynamic queues work to make it in :)

[doublewe]

I think setting the maximum concurrency for different queues is reasonable, as not all queues have the same scenario and processing time. I hope to merge the PRs as soon as possible. Thank you very much.

[kanzihuang]

@kanzihuang I assume this is still a WIP because I don't see the queue concurrency value being used in the processor/semaphore. I assume that is the end goal?

@kamikazechaser This is not a WIP. Here is the test code:
QueueConcurrency: map[string]int{base.DefaultQueueName: tc.queueConcurrency}

The following is the specific implementation code:
rdb := rdb.NewRDB(c, rdb.WithQueueConcurrency(cfg.QueueConcurrency))
func (r *RDB) EnqueueUnique ARGV[3]