Contexto:

Tenemos un microservicio que consume (suscribe) mensajes de más de 50 colas de RabbitMQ.

La producción de mensajes para esta cola ocurre en dos lugares

1. El proceso de la aplicación cuando encuentra una lógica comercial de ejecución retrasada (como enviar correos electrónicos O notificar a otro servicio), la aplicación envía directamente el mensaje al intercambio (que a su vez se envía a la cola).

2. Cuando nos encontramos con una lógica empresarial de ejecución larga/retrasada, tenemos una tabla de messages que tiene entradas de mensajes que deben ejecutarse después de un tiempo.

Ahora tenemos un trabajador cron que se ejecuta cada 10 minutos, escanea la tabla de messages y envía los mensajes a RabbitMQ.

Guión:

Digamos que la tabla de mensajes tiene 10 000 mensajes que se pondrán en cola en la próxima ejecución del cron,

1. 9:00 a. m.: se ejecuta Cron Worker y pone en cola 10 000 mensajes en la cola de RabbitMQ.
2. Tenemos suscriptores que escuchan la cola y comienzan a consumir los mensajes, pero debido a algún problema en el sistema o a un retraso en el tiempo de respuesta de terceros, cada mensaje tarda 1 Min en completarse.
3. 9:10 a. m.: ahora, el trabajador cron vuelve a ejecutar los próximos 10 minutos y observa que todavía hay más de 9000 mensajes por completar y también se ha cruzado el tiempo, por lo que una vez más envía más de 9000 mensajes duplicados a la cola.

Nota: los suscriptores que consumen los mensajes son idempotentes, por lo que no hay problema en el procesamiento duplicado

Idea de diseño que tenía en mente pero no la mejor lógica

Puedo tener 4 estados (RequiresQueuing, Queued, Completed, Failed)

1. Cada vez que se inserta un mensaje, puedo establecer el estado en RequiresQueuing
2. A continuación, cuando el trabajador cron selecciona y envía los mensajes con éxito a la cola, puedo configurarlo en Queued
3. Cuando los suscriptores completan, marcan el estado de la cola como Completed / Failed .

Hay un problema con la lógica anterior, digamos que RabbitMQ de alguna manera se cae O en algún uso hemos purgado la cola para el mantenimiento.

Ahora los mensajes que están marcados como En Queued están en un estado incorrecto, porque deben identificarse nuevamente y el estado debe cambiarse manualmente.

Otro ejemplo

Digamos que tengo el nombre RabbitMQ Queue (eventos)

Esta cola de eventos tiene 5 suscriptores, cada uno de los suscriptores recibe 1 mensaje de la cola y publica este evento utilizando la API REST en otro microservicio (agregador de eventos). Cada llamada a la API suele tardar 50 ms.

Caso de uso:

1. Debido a la alta carga, el número de eventos producidos se vuelve 3x.
2. Además, el microservicio (agregador de eventos) que acepta el evento también se volvió lento en el procesamiento, el tiempo de respuesta aumentó de 50 ms a 1 minuto.
3. Los trabajadores de Cron siguen su diseño mencionado anteriormente y ponen en cola el mensaje para cada minuto. Ahora la cola se está volviendo demasiado grande, pero tampoco puedo aumentar la cantidad de suscriptores porque el microservicio dependiente (agregador de eventos) también está retrasado.

Ahora la pregunta es: si sigue enviando los mensajes a la cola de eventos, solo está inflando la cola.

https://www.rabbitmq.com/memory.html : mientras leía esta página, descubrí que rabbitmq ni siquiera aceptará la conexión si alcanza una fracción de marca de agua alta (el valor predeterminado es 40%). Por supuesto, esto se puede cambiar, pero esto requiere una intervención manual.

Entonces, si la longitud de la cola aumenta, afecta la memoria de rabbitmq, esa es la razón por la que pensé en acelerar a nivel del productor.

Preguntas

1. ¿Cómo puedo acelerar mi trabajador cron para omitir esa ejecución en particular o de alguna manera inspeccionar la cola e identificar que ya está muy cargada para no enviar los mensajes?
2. ¿Cómo puedo manejar los casos de uso que dije anteriormente? ¿Existe un diseño que resuelva mi problema? ¿Alguien se enfrentó al mismo problema?

Gracias por adelantado.

Responder

Verifique la respuesta aceptada Comentarios para la limitación usando queueCount