Una cola de mensajes retardados (Delayed Queue) permite que los elementos no se procesen inmediatamente después de ser insertados, sino tras un periodo de tiempo específico. A diferencia de las colas convencionales donde el procesamiento sigue un orden FIFO (primero en entrar, primero en salir), aquí el factor determinante es la marca de tiempo programada.

Casos de Uso Comunes

Este patrón es fundamental en escenarios donde la ejecución debe posponerse de forma controlada. Por ejemplo:

• Notificaciones push: Enviar un recordatorio al usuario cinco minutos después de que se haya actualizado un contenido.
• E-commerce: Cancelar automáticamente un pedido si el pago no se recibe en 30 minutos.
• Reintentos con retroceso (backoff): Reprocesar una tarea fallida después de un intervalo creciente.

Arquitectrua basada en Redis ZSET

La implementación más eficiente en Redis utiliza el tipo de dato ZSET (Sorted Set). En esta estructura, cada elemento tiene asociado un "score" (puntaje). Para una cola de retardo, el score representa el timestamp Unix en el que el mensaje debe ser procesado.

A continuación, se define la interfaz base para la gestión de esta estructura:

public interface GestionadorColaRetrasada<T> {
    /**
     * Inserta un elemento con un tiempo de ejecución programado.
     * @param timestamp Tiempo en milisegundos para la ejecución.
     * @param payload Datos del mensaje.
     */
    boolean programarTarea(long timestamp, T payload);

    /**
     * Recupera el siguiente elemento disponible para ser procesado.
     */
    T extraerSiguiente();

    /**
     * Remueve un elemento de la cola.
     */
    boolean cancelarTarea(T payload);
}

Implementación con RedisTemplate

En el entorno de Spring Data Redis, la lógica se centra en utilizar opsForZSet. La clave para evitar condiciones de carrera en entornos distribuidos es asegurar que solo un consumidor procese el mensaje, lo cual logramos mediante la eliminación atómica del elemento (ZREM) inmediatamente después de leerlo.

public class RedisDelayedQueueImpl implements GestionadorColaRetrasada<String> {

    private final RedisTemplate<String, String> redisTemplate;
    private final String nombreCola;

    public RedisDelayedQueueImpl(String nombreCola, RedisTemplate<String, String> redisTemplate) {
        this.nombreCola = nombreCola;
        this.redisTemplate = redisTemplate;
    }

    @Override
    public boolean programarTarea(long timestamp, String payload) {
        return Boolean.TRUE.equals(redisTemplate.opsForZSet().add(nombreCola, payload, (double) timestamp));
    }

    @Override
    public String extraerSiguiente() {
        long ahora = System.currentTimeMillis();
        // Buscamos elementos cuyo score sea menor o igual al tiempo actual
        Set<String> items = redisTemplate.opsForZSet().rangeByScore(nombreCola, 0, ahora, 0, 1);
        
        if (items != null && !items.isEmpty()) {
            String valor = items.iterator().next();
            // Intentamos eliminar para garantizar exclusividad en el procesamiento
            if (Boolean.TRUE.equals(redisTemplate.opsForZSet().remove(nombreCola, valor) > 0)) {
                return valor;
            }
        }
        return null;
    }

    @Override
    public boolean cancelarTarea(String payload) {
        return redisTemplate.opsForZSet().remove(nombreCola, payload) > 0;
    }
}

Consumidor de Tareas (Worker Service)

Para procesar estos mensajes, se requiere un servicio en segundo plano que realice un sondeo (polling) constante sobre el conjunto ordenado. El uso de un pool de hilos permite escalar la capacidad de procesamiento.

public abstract class ProcesadorBaseCola implements InitializingBean {
    private static final Logger log = LoggerFactory.getLogger(ProcesadorBaseCola.class);

    private ExecutorService poolHilos;
    protected GestionadorColaRetrasada<String> colaObjetivo;
    private volatile boolean activo = true;

    public void configurar(GestionadorColaRetrasada<String> cola, int hilos) {
        this.colaObjetivo = cola;
        this.poolHilos = Executors.newFixedThreadPool(hilos);
    }

    @Override
    public void afterPropertiesSet() {
        poolHilos.execute(() -> {
            while (activo) {
                try {
                    String tarea = colaObjetivo.extraerSiguiente();
                    if (tarea != null) {
                        if (!ejecutarLogica(tarea)) {
                            gestionarFallo(tarea);
                        }
                    } else {
                        // Evitar consumo excesivo de CPU si la cola está vacía
                        Thread.sleep(1000);
                    }
                } catch (Exception e) {
                    log.error("Error en el ciclo del worker", e);
                }
            }
        });
    }

    /**
     * Define la lógica de negocio específica.
     */
    protected abstract boolean ejecutarLogica(String datos);

    protected void gestionarFallo(String datos) {
        log.warn("Tarea fallida: {}", datos);
    }

    public void detener() {
        this.activo = false;
        this.poolHilos.shutdown();
    }
}

Para calcular el tiempo de ejecución en la práctica, simplemente sumamos el retraso deseado al teimpo actual:

int minutosRetraso = 5;
long tiempoEjecucion = System.currentTimeMillis() + (minutosRetraso * 60 * 1000);
cola.programarTarea(tiempoEjecucion, "id_notificacion_123");

Este enfoque aprovecha la persistencia y la velocidad en memoria de Redis, ofreciendo una solución de colas retardadas robusta y fácil de implementar sin la complejidad de sistemas de mensajería pesados.