代码之家 › 专栏 › 技术社区 › wal-o-mat

Redis作为消息代理

redis

wal-o-mat · 技术社区 · 11 年前

问题

我希望以发布订阅的方式在应用程序之间传递数据。数据产生的速度可能比消耗的速度快得多,消息会丢失,这不是问题。想象一下一个快速传感器和一个慢速传感器数据处理器。为此,我使用redispub/sub并编写了一个类,该类充当订阅者,接收每个消息并将其放入缓冲区。当新消息传入时,缓冲区被覆盖或当“real”函数请求消息时无效。所以当我问这个类时,我会立即得到一个响应(提示我的函数比输入的数据慢)或者我必须等待(提示我函数比输入数据快)。

这对于数据输入速度快的情况非常有效。但对于相对较少的数据,比如说每5秒,这是不工作:想象一下我的消费者稍微启动了之后制作人,第一条消息丢失了,我的消费者需要等待近5秒,直到它开始工作。

我想我必须用Redis工具解决这个问题。而不是 发布/订阅 ,我可以简单地使用 获取/设置 方法,从而将缓存功能直接放入Redis。但是,我的消费者必须投票数据库而不是我现在拥有的事件魔法。密钥可能看起来像“密钥:时间戳”,我的消费者现在必须 get key:* 并永久比较时间戳,我认为这会导致大量负载。睡觉是不可能的,因为虽然我不在乎被丢弃的消息(我无能为力),但我确实在乎延迟。

是否有人使用Redis做类似的事情,并能给我一个关于如何巧妙使用Redis工具和数据结构的提示?

编辑

理想情况下,我的程序流程如下:

启动程序
检索 key 来自Redis
告诉Redis,“嘿,通知我 钥匙 ".
异步启动一些东西,并对新消息进行回调。

通过写这篇文章,一个想法出现了:出版商不仅出版 message 关于主题 钥匙 ,而且 set key message 这样,应用程序最初可以 get 然后 subscribe .

好主意还是不好?

我得到下面的答案(被接受的答案)后做了什么

密钥空间通知是我真正需要的。Redis作为信息的主要来源,我的客户订阅 键空间通知 ,其向订户通知影响特定密钥的事件。现在,在我的客户机的异步部分,我订阅有关我感兴趣的密钥的通知。这些通知设置了 key_has_updates 旗帜当我需要价值时,我 收到 并取消设置标志。使用未设置的标志,我知道服务器上该键没有新值。如果没有密钥空间通知,这将是我需要轮询服务器的部分。优点是我可以使用各种数据结构,而不仅仅是 pub/sub 机制,以及 慢接合工 它总是能够错过第一个事件 收到 初始值,其中 pub/sib 会丢失的。

当我需要该值时,我从Redis获取该值,并将标志设置为false。

3 回复 | 直到 11 年前

soveran 11 年前

一个想法是将数据推送到列表中( LPUSH )并修剪它( LTRIM )因此,如果没有消费者,它不会永远增长。另一方面,消费者将从该列表中获取项目并进行处理。您也可以使用 keyspace notifications ,并在每次将项目添加到该队列时发出警报。

Gianluca D'Ardia 9 年前

我使用两个本机redis命令在应用程序之间传递数据: rpush公司 和 blpop公司 . " blpop 当没有元素可从任何给定列表中弹出时,阻止连接”。

数据以json格式在应用程序之间传递,使用列表作为队列。
希望发送数据(充当发布者)的应用程序 rpush公司 在列表上
希望接收数据的应用程序(充当订阅者) blpop公司 在同一名单上

代码shuold be(用perl语言)

发件人 (我们假设一个哈希传递)

#Encode hash in json format
my $json_text = encode_json \%$hash_ref;

#Connect to redis and send to list
my $r = Redis->new(server => "127.0.0.1:6379");
$r->rpush("shared_queue","$json_text");
$r->quit;

接受者 (进入无限循环)

while (1) {
    my $r = Redis->new(server => "127.0.0.1:6379");
    my @elem =$r->blpop("shared_queue",0);

    #Decode hash element
    my $hash_ref=decode_json($elem\[1]);

    #make some stuff
}

我觉得这种方法非常有用,原因有很多:

该元素存储在列表中,因此临时禁用接收器不会丢失信息。当recevier重新启动时,可以处理列表中的所有项目。
可以使用多个接收方实例来处理高发送速率。
多个发件人可以在唯一列表中发送数据。在这种情况下,数据采集器应该很容易实现
可以使用特定工具(例如。 pm2 )

Nagendran 5 年前

在Redis5中,有一种新的数据类型叫做“Streams”,它是一种仅附加的数据结构。Redis流可以用作可靠的消息队列,使用消费者组概念进行点对点和多播通信 Redis_Streams_MQ