Tratando con la duplicación en una cola de mensajes

Question

He estado discutiendo con mi programador sobre la mejor manera de solucionar esto. Tenemos datos que llegan a una velocidad de aproximadamente 10000 objetos por segundo. Esto debe procesarse de forma asíncrona, pero la ordenación flexible es suficiente, por lo que cada objeto se inserta de manera rutinaria en una de varias colas de mensajes (también hay varios productores y consumidores). Cada objeto es ~ 300 bytes. Y debe ser duradero, por lo que los MQ están configurados para persistir en el disco.

El problema es que a menudo estos objetos están duplicados (ya que se duplican inevitablemente en los datos que llegan al productor). Tienen identificaciones únicas de 10 bytes. No es catastrófico si los objetos están duplicados en la cola, pero es si están duplicados en el procesamiento después de sacarlos de la cola. ¿Cuál es la mejor manera de asegurar lo más cerca posible de la escalabilidad lineal mientras se asegura que no haya duplicación en el procesamiento de los objetos? Y tal vez vinculado a eso, ¿debería almacenarse todo el objeto en la cola de mensajes, o solo el ID con el cuerpo almacenado en algo como cassandra?

Gracias!

Editar: Confirmado donde ocurre la duplicación. Además, hasta ahora he tenido 2 recomendaciones para Redis. Anteriormente había estado considerando RabbitMQ. ¿Cuáles son los pros y los contras de cada uno con respecto a mis requisitos?

Answer 1

Sin saber cómo se crean los mensajes dentro del sistema, el mecanismo que el productor utiliza para publicar en la cola y saber con el sistema de cola está en uso, es difícil diagnosticar lo que está sucediendo.

He visto este escenario suceder de varias maneras diferentes; trabajadores agotados que causan que el mensaje vuelva a ser visible en la cola (y así procesado por segunda vez, esto es común con Kestrel), corredores mal configurados (HA ActiveMQ viene a la mente), clientes mal configurados (el enrutamiento Spring plus Camel viene a la mente) , clientes que envían dos veces, etc. Solo hay varias formas en que puede surgir este tipo de problema.

Como realmente no puedo diagnosticar el problema, voy a conectar redis aquí. Podría combinar fácilmente algo como SPOP (que es O (1), como lo es SADD) con pub/sub para una cola increíblemente rápida, de tiempo constante, libre de duplicados (los conjuntos deben contener elementos únicos). Aunque es un proyecto de ruby, resque puede ayudar. Al menos vale la pena mirarlo.

Lo mejor de la suerte.

Answer 2

PS: esta es la primera vez en mi vida que ReDiS sitio web es tener problemas, pero apuesto a que cuando la visite, que han resuelto el problema

> We have data that comes in at a rate 
> of about 10000 objects per second. 
> This needs to be processed 
> asynchronously, but loose ordering is 
> sufficient, so each object is inserted 
> round-robin-ly into one of several 
> message queues (there are also several 
> producers and consumers) 

Mi primer consejo sería buscar al redis porque es increíblemente rápido y apuesto a que puedes manejar todos tus mensajes con solo una cola de mensajes.

Primero me gusta mostrarle información sobre mi computadora portátil (me gusta, pero un servidor grande va a ser mucho más rápido;)). Mi padre (quedó impresionado un poco :)) recientemente compró una nueva PC y es mucho mejor que mi computadora portátil (8 CPU en lugar de 2).

-Computer- 
Processor  : 2x Intel(R) Core(TM)2 Duo CPU  T7100 @ 1.80GHz 
Memory  : 2051MB (1152MB used) 
Operating System  : Ubuntu 10.10 
User Name  : alfred (alfred) 
-Display- 
Resolution  : 1920x1080 pixels 
OpenGL Renderer  : Unknown 
X11 Vendor  : The X.Org Foundation 
-Multimedia- 
Audio Adapter  : HDA-Intel - HDA Intel 
-Input Devices- 
Power Button 
Lid Switch 
Sleep Button 
Power Button 
AT Translated Set 2 keyboard 
Microsoft Comfort Curve Keyboard 2000 
Microsoft Comfort Curve Keyboard 2000 
Logitech Trackball 
Video Bus 
PS/2 Logitech Wheel Mouse 
-SCSI Disks- 
HL-DT-ST DVDRAM GSA-T20N 
ATA WDC WD1600BEVS-2 

Por debajo de los puntos de referencia utilizando redis-benchmark en mi máquina sin ni siquiera hacer Redis mucho optimización:

alfred@alfred-laptop:~/database/redis-2.2.0-rc4/src$ ./redis-benchmark 
====== PING (inline) ====== 
    10000 requests completed in 0.22 seconds 
    50 parallel clients 
    3 bytes payload 
    keep alive: 1 

94.84% <= 1 milliseconds 
98.74% <= 2 milliseconds 
99.65% <= 3 milliseconds 
100.00% <= 4 milliseconds 
46296.30 requests per second 

====== PING ====== 
    10000 requests completed in 0.22 seconds 
    50 parallel clients 
    3 bytes payload 
    keep alive: 1 

91.30% <= 1 milliseconds 
98.21% <= 2 milliseconds 
99.29% <= 3 milliseconds 
99.52% <= 4 milliseconds 
100.00% <= 4 milliseconds 
45662.10 requests per second 

====== MSET (10 keys) ====== 
    10000 requests completed in 0.32 seconds 
    50 parallel clients 
    3 bytes payload 
    keep alive: 1 

3.45% <= 1 milliseconds 
88.55% <= 2 milliseconds 
97.86% <= 3 milliseconds 
98.92% <= 4 milliseconds 
99.80% <= 5 milliseconds 
99.94% <= 6 milliseconds 
99.95% <= 9 milliseconds 
99.96% <= 10 milliseconds 
100.00% <= 10 milliseconds 
30864.20 requests per second 

====== SET ====== 
    10000 requests completed in 0.21 seconds 
    50 parallel clients 
    3 bytes payload 
    keep alive: 1 

92.45% <= 1 milliseconds 
98.78% <= 2 milliseconds 
99.00% <= 3 milliseconds 
99.01% <= 4 milliseconds 
99.53% <= 5 milliseconds 
100.00% <= 5 milliseconds 
47169.81 requests per second 

====== GET ====== 
    10000 requests completed in 0.21 seconds 
    50 parallel clients 
    3 bytes payload 
    keep alive: 1 

94.50% <= 1 milliseconds 
98.21% <= 2 milliseconds 
99.50% <= 3 milliseconds 
100.00% <= 3 milliseconds 
47619.05 requests per second 

====== INCR ====== 
    10000 requests completed in 0.23 seconds 
    50 parallel clients 
    3 bytes payload 
    keep alive: 1 

91.90% <= 1 milliseconds 
97.45% <= 2 milliseconds 
98.59% <= 3 milliseconds 
99.51% <= 10 milliseconds 
99.78% <= 11 milliseconds 
100.00% <= 11 milliseconds 
44444.45 requests per second 

====== LPUSH ====== 
    10000 requests completed in 0.21 seconds 
    50 parallel clients 
    3 bytes payload 
    keep alive: 1 

95.02% <= 1 milliseconds 
98.51% <= 2 milliseconds 
99.23% <= 3 milliseconds 
99.51% <= 5 milliseconds 
99.52% <= 6 milliseconds 
100.00% <= 6 milliseconds 
47619.05 requests per second 

====== LPOP ====== 
    10000 requests completed in 0.21 seconds 
    50 parallel clients 
    3 bytes payload 
    keep alive: 1 

95.89% <= 1 milliseconds 
98.69% <= 2 milliseconds 
98.96% <= 3 milliseconds 
99.51% <= 5 milliseconds 
99.98% <= 6 milliseconds 
100.00% <= 6 milliseconds 
47619.05 requests per second 

====== SADD ====== 
    10000 requests completed in 0.22 seconds 
    50 parallel clients 
    3 bytes payload 
    keep alive: 1 

91.08% <= 1 milliseconds 
97.79% <= 2 milliseconds 
98.61% <= 3 milliseconds 
99.25% <= 4 milliseconds 
99.51% <= 5 milliseconds 
99.81% <= 6 milliseconds 
100.00% <= 6 milliseconds 
45454.55 requests per second 

====== SPOP ====== 
    10000 requests completed in 0.22 seconds 
    50 parallel clients 
    3 bytes payload 
    keep alive: 1 

91.88% <= 1 milliseconds 
98.64% <= 2 milliseconds 
99.09% <= 3 milliseconds 
99.40% <= 4 milliseconds 
99.48% <= 5 milliseconds 
99.60% <= 6 milliseconds 
99.98% <= 11 milliseconds 
100.00% <= 11 milliseconds 
46296.30 requests per second 

====== LPUSH (again, in order to bench LRANGE) ====== 
    10000 requests completed in 0.23 seconds 
    50 parallel clients 
    3 bytes payload 
    keep alive: 1 

91.00% <= 1 milliseconds 
97.82% <= 2 milliseconds 
99.01% <= 3 milliseconds 
99.56% <= 4 milliseconds 
99.73% <= 5 milliseconds 
99.77% <= 7 milliseconds 
100.00% <= 7 milliseconds 
44247.79 requests per second 

====== LRANGE (first 100 elements) ====== 
    10000 requests completed in 0.39 seconds 
    50 parallel clients 
    3 bytes payload 
    keep alive: 1 

6.24% <= 1 milliseconds 
75.78% <= 2 milliseconds 
93.69% <= 3 milliseconds 
97.29% <= 4 milliseconds 
98.74% <= 5 milliseconds 
99.45% <= 6 milliseconds 
99.52% <= 7 milliseconds 
99.93% <= 8 milliseconds 
100.00% <= 8 milliseconds 
25906.74 requests per second 

====== LRANGE (first 300 elements) ====== 
    10000 requests completed in 0.78 seconds 
    50 parallel clients 
    3 bytes payload 
    keep alive: 1 

1.30% <= 1 milliseconds 
5.07% <= 2 milliseconds 
36.42% <= 3 milliseconds 
72.75% <= 4 milliseconds 
93.26% <= 5 milliseconds 
97.36% <= 6 milliseconds 
98.72% <= 7 milliseconds 
99.35% <= 8 milliseconds 
100.00% <= 8 milliseconds 
12886.60 requests per second 

====== LRANGE (first 450 elements) ====== 
    10000 requests completed in 1.10 seconds 
    50 parallel clients 
    3 bytes payload 
    keep alive: 1 

0.67% <= 1 milliseconds 
3.64% <= 2 milliseconds 
8.01% <= 3 milliseconds 
23.59% <= 4 milliseconds 
56.69% <= 5 milliseconds 
76.34% <= 6 milliseconds 
90.00% <= 7 milliseconds 
96.92% <= 8 milliseconds 
98.55% <= 9 milliseconds 
99.06% <= 10 milliseconds 
99.53% <= 11 milliseconds 
100.00% <= 11 milliseconds 
9066.18 requests per second 

====== LRANGE (first 600 elements) ====== 
    10000 requests completed in 1.48 seconds 
    50 parallel clients 
    3 bytes payload 
    keep alive: 1 

0.85% <= 1 milliseconds 
9.23% <= 2 milliseconds 
11.03% <= 3 milliseconds 
15.94% <= 4 milliseconds 
27.55% <= 5 milliseconds 
41.10% <= 6 milliseconds 
56.23% <= 7 milliseconds 
78.41% <= 8 milliseconds 
87.37% <= 9 milliseconds 
92.81% <= 10 milliseconds 
95.10% <= 11 milliseconds 
97.03% <= 12 milliseconds 
98.46% <= 13 milliseconds 
99.05% <= 14 milliseconds 
99.37% <= 15 milliseconds 
99.40% <= 17 milliseconds 
99.67% <= 18 milliseconds 
99.81% <= 19 milliseconds 
99.97% <= 20 milliseconds 
100.00% <= 20 milliseconds 
6752.19 requests per second 

Como se puede ver con suerte de la evaluación comparativa de mi portátil sencilla es probable que sólo necesita una cola de mensajes porque Redis puede manejar 10000 lpush solicitudes en 0,23 segundos y 10000 lpop solicitudes en 0,21 segundos. Cuando sólo se necesita una cola Creo que su problema no es ya un problema (o son los productores producings duplicados, que no entiendo por completo?).

> And it needs to be durable, so the MQs 
> are configured to persist to disk. 

redis también persisten en el disco.

> The problem is that often these 
> objects are duplicated. They do have 
> 10-byte unique ids. It's not 
> catastrophic if objects are duplicated 
> in the queue, but it is if they're 
> duplicated in the processing after 
> being taken from the queue. What's the 
> best way to go about ensuring as close 
> as possible to linear scalability 
> whilst ensuring there's no duplication 
> in the processing of the objects? 

Al usar una sola cola de mensajes (recuadro) este problema no existe si entiendo correctamente. Pero si no, simplemente podría verificar si el ID is member of your set ids. Cuando procesa la identificación debe remove it from the set ids. En primer lugar se debe añadir offcourse los miembros de la lista mediante sadd.

Si una caja no escala ya que debe fragmentar sus claves a través de múltiples cajas y comprobar esa tecla en esa caja. para aprender más sobre esto creo que usted debe leer los siguientes enlaces:

• Redis replication and redis sharding (cluster) difference
• http://antirez.com/post/redis-presharding.html
• http://redis.io/presentation/Redis_Cluster.pdf

• http://blog.zawodny.com/2011/02/26/redis-sharding-at-craigslist/

  Y tal vez vinculado a que, si el conjunto objeto se almacena en la cola de mensajes , o sólo el ID con el cuerpo almacenada en algo como Cassandra?

Si es posible, debe toda su información directamente en la memoria porque nada puede correr tan rápido como la memoria (bueno su memoria cache es incluso más rápido, pero muy, muy pequeña además de que no se puede acceder a través de que su código) . Redis almacena toda su información dentro de la memoria y hace instantáneas en el disco. Creo que deberías ser capaz de almacenar toda tu información en la memoria y saltarte usando algo así como Cassandra.

Consideremos que cada objeto tiene 400 bytes por objeto en total a una velocidad de 10000 por segundo => 4000000 bytes para todos los objetos por segundo => 4 MB/s si mi cálculo es correcto. Puede almacenar fácilmente esa cantidad de información dentro de su memoria. Si no puedes, deberías considerar actualizar tu memoria si es posible, porque la memoria ya no es tan cara.

Answer 3

Si no te importa tirar Camel en la mezcla, entonces puedes usar el EIP idempotent-consumer para ayudarte con esto.

Además, ActiveMQ Message Groups se puede utilizar para mensajes relacionados con el grupo y hacerlos más fáciles de realizar comprobaciones de duplicados y todavía mantienen un alto rendimiento, etc ...