处理消息队列中的重复问题

我一直在和我的程序员讨论如何解决这个问题。我们有数据以每秒大约10000个对象的速度进入。这需要异步处理，但松散排序就足够了，因此每个对象都被循环插入到几个消息队列中（也有几个生产者和消费者）。每个对象都是〜300字节。它需要持久耐用，所以MQ被配置为坚持磁盘。处理消息队列中的重复问题

问题是，这些对象经常被复制（因为它们不可避免地会在生产者的数据中被复制）。他们有10个字节的唯一ID。如果对象在队列中被复制，这不是灾难性的，但是如果它们在从队列中取出之后在处理中被复制，那就不是灾难性的。什么是确保尽可能接近线性可扩展性同时确保对象处理没有重复的最佳方法？也许与此有关，应该将整个对象存储在消息队列中，还是只将存储在cassandra中的身份标识存储在内？

谢谢！

编辑：确认发生重复。另外，到目前为止，我对Redis有2条建议。我以前一直在考虑RabbitMQ。关于我的要求，每种方式的优缺点是什么？

来源

2011-06-28 Max

Redis的是非常快的开源，先进的key-value存储。它通常被称为数据结构服务器，因为密钥可以包含字符串，哈希，列表，集合和有序集合。 Redis也有（非常）积极的开发，如果你问我，很有用。我从来没有玩过RabbitMQ。 – Alfred

我将添加另一个Redis插件。我最近开发了一个[开源消息队列项目]（http://jordanhalterman.github.com/redmq/），该项目最初以另一种存储方法开始（保持未命名状态）。在项目中途，我决定切换到Redis。我发现它非常快速，非常可靠，易于学习，并且具有构建基本和复杂消息系统的非常有用的功能。 – kuujo

不知道如何在系统内创建消息，生产者用于发布到队列中的机制，并且知道队列系统正在使用中，因此很难诊断正在发生的事情。

我见过这种情况发生在许多不同的方式;导致消息在队列中再次可见（并且因此第二次处理，这在Kestrel中很常见），错误配置的代理（HA ActiveMQ浮现），错误配置的客户端（想到Spring和Camel路由），客户双提交等等。这种问题出现的方式有很多种。

由于我无法真正诊断问题，因此我会在此处插入redis。你可以很容易地将SPOP（这是O（1），就像是SADD）和pub/sub这样的一个快速，持续时间，重复空闲（集合必须包含唯一元素）队列。虽然这是一个红宝石项目，但resque也许能够提供帮助。至少值得一看。

祝你好运。

来源

2011-06-28 03:15:29 bmatheny

感谢您对redis @bmatheny的建议！只是通过结合SPOP和pub/sub来检查你是否指的是生产者SADD，然后发布给消费者以通知他们增加的内容？另外，如果我将SADDing设置为ID，传达其余对象数据的最佳方式是什么？ – Max

取决于你想如何实现它。如果你不介意有NOOP消费者（消费者没有工作，这是一个民意调查模型），你可以为消费者/生产者使用SPOP和SADD。您也可以使用BLPOP（它将阻塞，直到数据可用），但仅适用于可能有重复的列表。发布/订阅的实现与列表无关，因此可能会有欺骗。如果你想消除愚蠢，坚持套和投票的消费者。 – bmatheny

PS：这是我人生中第一次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 cpu的，而不是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

正如你希望从基准我简单的笔记本电脑，你可能只需要一个消息队列，因为可以Redis的处理看10000个lpush请求0.23秒，在0.21秒10000个lpop请求。当你只需要一个队列时，我相信你的问题已经不再是问题了（或者制作人制作的复制品我完全不理解？）。

> 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'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?

当使用单个消息队列（框）时，如果我理解正确，则不存在此问题。但是，如果不是，你可以只是简单地检查是否is member of your set ids。当你处理ID时，你应该remove it from the set ids。首先，您应该使用sadd将成员添加到列表中。

如果一个框不会再扩展你应该分片你的钥匙在多个箱子和检查上框键。要了解更多关于这一点，我想你应该阅读下面的链接：

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/

或许链接到，应全对象存储在消息队列中，或者只有身份为0的id存储在像cassandra的东西？

如果可能的话，你应该所有的信息直接到内存中，因为没有什么可以尽可能快地运行内存（好吧你cache内存还要快，但真的真的很小，再加上你不能访问通过您的代码）。 Redis确实将所有信息存储在内存中，并将快照制作成光盘。我认为你应该能够将所有的信息存储在内存中，并且完全跳过像Cassandra这样的东西。

让我们考虑每个对象是合计每个对象400个字节以10000每秒=> 4000000字节每秒=> 4百万字节/秒，如果我的计算是正确的所有对象的速率。您可以轻松地将这些信息存储在内存中。如果你不能真的考虑升级你的内存，如果可能的话，因为内存不再那么昂贵。

来源

2011-06-28 05:34:56 Alfred

如果您不介意将Camel投入组合，那么您可以使用EIP来帮助解决这个问题。

此外，ActiveMQ Message Groups可用于组相关的消息，使他们更容易进行重复检查，并仍然保持高吞吐量，等等

来源

2011-06-29 23:54:03

处理消息队列中的重复问题

回答

相关问题