zlacker

The way I implement my queues (usually as part of my monolith application) is as go routines. Each instance of the app launches with a unique id, and also a role. It can be a worker or the app itself. So when the app generates a queue item, it simply adds it to a table as pending. A worker will then, via transaction, update a set of items to add its instance id as well as an expiration for this lock. If that succeeds, no other worker will pull a queue item with a non null id or with an id different that it’s instance id that is not expired. Worker can then start processing and update item status accordingly. If it crashes, another worker will just repeat the process after the lock expires.

The code that does this is maybe 100 lines at most. It’s very effective especially if you deploy your app in kubernetes where you can expect instances to be ephemeral. It’s one of the components of my apps that has never needed any updates since I first wrote it circa 2017.

>>figass+(OP)
How are workers notified of new work? Or do they poll and sleep?

>>perfmo+R
They poll. It's easy however to add pub/sub via redis. Queues are categorized in frequency. So for example a notification queue might process every minute or 5 minutes, a reconciliation queue might run every 12 hours, or have a fixed time, like daily at 2am and 6pm, etc. Each queue runs on its own goroutine on its onwn schedule, and every worker runs all queues. I can also adjust how many items a worker can pull each time so queues do not get backed up. Say there is some criteria an account needs to meet to be eligible for some feature. If that criteria requires some expensive db queries, I can run a daily job that pulls 100 accounts each time to check that they meet the criteria. But if the instances can process more, I can configure it to pull 1000 accounts. Or I can add 9 workers and each pulls 100 accounts.

Usually what I do is pull the records that were updated least recently (as in they should be ahead of the queue). So if a previous worker locked the oldest X records, the second worker will pull the next batch bc the condition will exclude the previously updated (locked) records. There's a lot of flexibility you can add with just these controls: schedule, frequency, batch size, number of workers.