Работа с дублированием в очереди сообщений

Я спорил с моим программистом о лучшем способе этого. У нас есть данные, которые поступают со скоростью около 10000 объектов в секунду. Это нужно обрабатывать асинхронно, но свободного упорядочения достаточно, поэтому каждый объект вставляется в одно из нескольких очередей сообщений (есть также несколько производителей и потребителей). Каждый объект ~ 300 байт. И это должно быть долговечным, поэтому MQ настроены на сохранение на диске.

Проблема в том, что часто эти объекты дублируются (так как они неизбежно дублируются в данных, поступающих к производителю). У них есть 10-байтовые уникальные идентификаторы. Это не катастрофично, если объекты дублируются в очереди, но это происходит, если они дублируются в обработке после того, как они были взяты из очереди. Каков наилучший способ обеспечить как можно более близкую к линейной масштабируемости, не обеспечивая дублирования при обработке объектов? И, возможно, связанный с этим, должен ли весь объект храниться в очереди сообщений или только идентификатор с телом, хранящимся в чем-то вроде cassandra?

Спасибо!

Изменить: Подтверждено, где происходит дублирование. Кроме того, до сих пор у меня было 2 рекомендации для Redis. Раньше я рассматривал RabbitMQ. Каковы плюсы и минусы каждого из моих требований?

Ответ 1

Не зная, как создаются сообщения в системе, механизм, который производитель использует для публикации в очереди, и знание системы с очередями используется, трудно диагностировать, что происходит.

Я видел, как этот сценарий происходит по-разному; (и, таким образом, он обрабатывается во второй раз, это часто встречается с Kestrel), неправильно сконфигурированные брокерами (на ум приходит HA ActiveMQ), неправильно сконфигурированные клиенты (Spring плюс маршрутизация Camel приходит к ум), клиенты с двойным представлением и т.д. Существует всего несколько способов, чтобы такая проблема могла возникать.

Так как я не могу действительно диагностировать проблему, я подключу redis здесь. Вы можете легко комбинировать что-то вроде SPOP (это O (1), как SADD) с pub/sub для невероятно быстрого, постоянного времени, дублирования свободных (набор должен содержать уникальные элементы) очереди. Хотя это рубиновый проект, resque может помочь. Это, по крайней мере, стоит посмотреть.

Желаем удачи.

Ответ 2

p.s: это первый раз в моей жизни, когда на веб-сайте redis возникают проблемы, но я готов поспорить, когда вы его посещаете, они решили проблему

> 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)

Мой первый совет - посмотреть redis, потому что это безумно быстро, и я уверен, вы можете обрабатывать все свои сообщения только одна очередь сообщений.

Сначала я хотел бы показать вам информацию о моем ноутбуке (мне это нравится, но большой сервер будет намного быстрее;)). Мой папа (был немного впечатлен:)) недавно купил новый компьютер, и он сильно бьет мой ноутбук (вместо 8 процессоров вместо 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

Ниже контрольных показателей, использующих redis-benchmark на моей машине, даже не делая большой оптимизации redis:

[email protected]:~/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

Как вы можете надеяться увидеть, сравнивая мой простой ноутбук, вам, вероятно, просто нужна одна очередь сообщений, потому что может перерисовать дескриптор 10000 lpush в 0.23 секунд и 10000 lpop запрашивает через 0,21 секунды. Когда вам просто нужна одна очередь, я считаю, что ваша проблема больше не проблема (или дубликаты продюсеров, которые я не понимаю полностью?).

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

redis также сохраняется на диске.

> The problem is that often these
> objects are duplicated. They do have
> 10-byte unique ids. It 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 the
> best way to go about ensuring as close
> as possible to linear scalability
> whilst ensuring there no duplication
> in the processing of the objects?

При использовании одной очереди сообщений (в поле) эта проблема не существует, если я правильно понимаю. Но если бы вы не могли просто проверить, является ли id членом ваших идентификаторов set. Когда вы обрабатываете идентификатор, вы должны удалить его из установленных идентификаторов. Сначала вы должны отказаться от добавления членов в список, используя sadd.

Если одно окно больше не масштабируется, вы должны очертить свои ключи на нескольких ящиках и проверить этот ключ на этом поле. Чтобы узнать больше об этом, я думаю, вы должны прочитать следующие ссылки:

Редизайн Redis и перерасчет изменений (кластеров)
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/

И, возможно, связанный с этим, если весь объект будет сохранен в сообщении очередь или только идентификатор с телом хранится в чем-то вроде cassandra?

Если возможно, вы должны использовать всю свою информацию непосредственно в памяти, потому что ничто не может работать так быстро, как память (хорошо, что ваша cache память даже быстрее, но действительно очень мало плюс вы не можете получить доступ к этому через свой код). Redis сохраняет всю вашу информацию в памяти и делает снимки на диск. Я думаю, вы должны иметь возможность хранить всю свою информацию в памяти и пропустить что-то вроде Cassandra.

Рассмотрим, что каждый объект составляет 400 байт на объект в общей сложности со скоростью 10000 в секунду = > 4000000 байт для всех объектов в секунду = > 4 МБ/с, если мои вычисления верны. Вы можете легко сохранить этот объем информации в своей памяти. Если вы не можете, вам стоит подумать об обновлении вашей памяти, если это вообще возможно, потому что память уже не так дорого.

Ответ 3

Если вы не против бросать Camel в микс, вы можете использовать idempotent-consumer EIP, чтобы помочь с этим.

Кроме того, Группы сообщений ActiveMQ можно использовать для группировки связанных сообщений и упрощения выполнения дублирующих проверок и поддержания высокой пропускной способности и т.д...