Redis Pub/Sub, BullMQ Job Queues, and Rate Limiting

Redis Beyond Caching

Redis excels as a message broker, job queue, and rate limiter.

BullMQ Job Queue

import { Queue, Worker } from 'bullmq'
const queue = new Queue('email', { connection })

await queue.add('welcome', { to: 'user@example.com' }, {
  attempts: 3,
  backoff: { type: 'exponential', delay: 1000 },
  removeOnComplete: 100,
})

// Cron job
await queue.add('digest', {}, { repeat: { cron: '0 9 * * MON' } })

// Worker
const worker = new Worker('email', async (job) => {
  await sendEmail(job.data)
}, { connection, concurrency: 10 })

worker.on('failed', (job, err) => console.error('Job failed:', job?.id, err))

Pub/Sub

await subscriber.subscribe('notifications', (msg) => {
  console.log('Got:', JSON.parse(msg))
})
await publisher.publish('notifications', JSON.stringify({ type: 'order_shipped', id: '123' }))

Sliding Window Rate Limiter (Lua)

const script = `
local key = KEYS[1]
local limit, now, window = tonumber(ARGV[1]), tonumber(ARGV[2]), tonumber(ARGV[3])
redis.call('ZREMRANGEBYSCORE', key, 0, now - window * 1000)
local count = redis.call('ZCARD', key)
if count >= limit then return 0 end
redis.call('ZADD', key, now, now)
redis.call('EXPIRE', key, window)
return 1
`

async function rateLimit(userId, limit, windowSecs) {
  return await redis.eval(script, {
    keys: ['rate:' + userId],
    arguments: [String(limit), String(Date.now()), String(windowSecs)]
  })
}

-> Generate API keys with the Token Generator.