Bloqueos y mensajes de recuperación de lotes con RabbitMq

Question

Estoy tratando de usar RabbitMq de una manera menos convencional (aunque en este punto puedo elegir cualquier otra implementación de cola de mensajes si es necesario). En lugar de dejar los mensajes push de Rabbit a mis consumidores, el consumidor se conecta a una cola y obtiene un lote de N mensajes (durante los cuales consume algunos y rechaza algunos), luego pasa a otra fila, y así sucesivamente. Esto se hace por redundancia. Si algunos consumidores se cuelgan, se garantiza que todos los mensajes serán consumidos por otro consumidor.

El problema es que tengo varios consumidores y no quiero que compitan en la misma fila. ¿Hay alguna manera de garantizar un bloqueo en una cola? Si no, ¿puedo al menos asegurarme de que si 2 consumidores están conectados a la misma cola, no leen el mismo mensaje? Las transacciones pueden ayudarme hasta cierto punto, pero he escuchado que se eliminarán de RabbitMQ.

También se aceptan otras sugerencias arquitectónicas.

Gracias!

EDITAR: Como se señala en el comentario, hay una particularidad en la forma en que necesito procesar los mensajes. Solo tienen sentido tomarlos en grupos y hay una gran probabilidad de que los mensajes relacionados se agrupen en una cola. Si, por ejemplo, saco un lote de 100 mensajes, hay una gran probabilidad de que pueda hacer algo con los mensajes 1-3, 4-5, 6-10, etc. Si no encuentro un grupo para algunos mensajes, Los volveré a enviar a la cola. WorkQueue no funcionaría porque difundiría mensajes del mismo grupo a varios trabajadores que no sabrían qué hacer con ellos.

Answer 1

¿Has echado un vistazo a este libro en línea gratis en Enterprise Integration Patterns?

Parece que realmente necesita un flujo de trabajo donde tenga un componente de dosificador antes de que los mensajes lleguen a sus trabajadores. Con RabbitMQ hay dos formas de hacerlo. Utilice un tipo de intercambio (y formato de mensaje) que pueda realizar el procesamiento por lotes, o tenga una cola, y un trabajador que clasifique los lotes y coloque cada lote en su propia cola. Es probable que el dosificador también envíe un mensaje de "lote listo" a una cola de control para que un trabajador pueda descubrir la existencia de la nueva cola de lote. Una vez que se procesa el lote, el trabajador puede eliminar la cola del lote.

Si tiene control sobre el formato del mensaje, es posible que pueda obtener RabbitMQ para realizar los lotes implícitamente de varias maneras. Con un intercambio de temas, puede asegurarse de que la clave de enrutamiento de cada mensaje tenga el formato work.batchid.something y luego un trabajador que sepa de la existencia de batch xxyzz use una clave de enlace como # .xxyzz. # Para solo consume esos mensajes. No es necesario republicar.

La otra forma es incluir un id. De lote en un encabezado y usar el tipo de intercambio de encabezados más nuevo. Por supuesto, también puede implementar sus propios tipos de intercambio personalizados si está dispuesto a escribir una pequeña cantidad de código Erlang.

Sin embargo, recomiendo consultar el libro, ya que ofrece una mejor visión general de la arquitectura de mensajería que el concepto típico de cola de trabajo con el que la mayoría de la gente comienza.

Answer 2

Haga que sus consumidores saquen de una sola cola. Se les garantizará que no compartan mensajes (Rabbit hará un intercambio de mensajes entre los consumidores actualmente conectados) y está muy optimizado para ese patrón de uso exacto.

Está listo para usar, listo para utilizar. En los documentos RabbitMQ se llama modelo Work Queue. Una cola, múltiples consumidores, sin que ninguno de ellos comparta nada. Suena como lo que necesitas.

Answer 3

Puede establecer un recuento de captación previa de nivel de canal/consumidor para consumir mensajes en lotes. Para volver a enviar los mensajes, debe usar el método AMQP basic.reject y esos mensajes se pueden elegir para ponerlos en cola o reenviarlos a la cola de mensajes no entregados. Múltiples consumidores que intentan extraer mensajes de la misma cola no es un problema ya que el método AMQP basic.get se sincronizará para manejar a los consumidores concurrentes.

https://groups.google.com/forum/#!topic/rabbitmq-users/hJ8f5du-GCA