php 69 lines · 8 steps

Deduplicating a job queue with Redis

A dispatcher uses an atomic Redis SET NX to ensure each unique job is enqueued only once within a TTL window.

Explained by highlit

Advanced deduplication redis atomicity hashing queueing

1<?php
2 
3namespace App\Queue;
4 
5use Predis\Client as Redis;
6 
7final class DeduplicatingDispatcher
8{
9    private const DEDUP_TTL = 3600;
10 
11    public function __construct(
12        private readonly Redis $redis,
13        private readonly JobQueue $queue,
14    ) {
15    }
16 
17    public function dispatch(string $type, array $payload): ?string
18    {
19        $fingerprint = $this->fingerprint($type, $payload);
20        $dedupKey = "queue:dedup:{$fingerprint}";
21 
22        $acquired = $this->redis->set($dedupKey, '1', 'EX', self::DEDUP_TTL, 'NX');
23 
24        if ($acquired === null) {
25            return null;
26        }
27 
28        try {
29            $jobId = $this->queue->push([
30                'type' => $type,
31                'payload' => $payload,
32                'fingerprint' => $fingerprint,
33            ]);
34        } catch (\Throwable $e) {
35            $this->redis->del([$dedupKey]);
36            throw $e;
37        }
38 
39        return $jobId;
40    }
41 
42    public function release(string $fingerprint): void
43    {
44        $this->redis->del(["queue:dedup:{$fingerprint}"]);
45    }
46 
47    private function fingerprint(string $type, array $payload): string
48    {
49        $this->ksortRecursive($payload);
50 
51        $canonical = json_encode(
52            ['type' => $type, 'payload' => $payload],
53            JSON_THROW_ON_ERROR | JSON_UNESCAPED_SLASHES,
54        );
55 
56        return hash('sha256', $canonical);
57    }
58 
59    private function ksortRecursive(array &$data): void
60    {
61        foreach ($data as &$value) {
62            if (is_array($value)) {
63                $this->ksortRecursive($value);
64            }
65        }
66 
67        ksort($data);
68    }
69}

01 / 01

STEP 01

Walkthrough

Space play ←→ step click any line

Three takeaways

1An atomic SET with NX and EX turns Redis into a self-expiring lock that guards against duplicate work.
2Canonicalizing input before hashing ensures logically identical payloads produce the same fingerprint.
3Releasing the dedup key on failure keeps a transient error from permanently blocking a legitimate retry.

Related explainers

php

namespace App\Providers;
 
use Illuminate\Support\Facades\Blade;
use Illuminate\Support\ServiceProvider;

Building a @money Blade directive in Laravel

blade-directive service-provider localization

Intermediate 5 steps

php

<?php
 
namespace App\Providers;

Defining authorization gates in Laravel

authorization gates service-provider

Intermediate 6 steps

javascript

import { NextResponse } from 'next/server';
import { Redis } from '@upstash/redis';
 
const redis = Redis.fromEnv();

Sliding-window rate limiting in a Next.js route

rate-limiting redis sorted-set

Advanced 8 steps

php

<?php
 
final class TaskProgressReporter
{

Tracking task progress in a JSON file

state-persistence atomic-writes json

Intermediate 9 steps

python

import pandas as pd
import numpy as np

Cleaning a customer DataFrame with pandas

data-cleaning regex normalization

Intermediate 9 steps

php

<?php
 
namespace App\Http\Requests\DataObjects;

Typed request DTOs in Laravel

data-transfer-object validation immutability

Intermediate 6 steps

Share this explainer

Here's the card — post it anywhere.

Deduplicating a job queue with Redis — share card

Made with highlit — turn any snippet into a walkthrough like this in about a minute.

Explain your code