使用 Redis 流实现消息队列的代码

在介绍了 Redis 流的基本功能之后, 现在是时候使用这些功能来构建一些实际的应用了。 消息队列作为流的典型应用之一, 具有非常好的示范性, 因此我们将使用 Redis 流的相关功能构建一个消息队列应用, 这个消息队列跟我们之前使用其他 Redis 数据结构构建的消息队列具有相似的功能。

代码清单 10-1 展示了一个具有基本功能的消息队列实现:

  • 代码最开头的是几个转换函数, 它们负责对程序的相关输入输出进行转换和格式化;
  • MessageQueue 类用于实现消息队列, 它的添加消息、移除消息以及返回消息数量三个方法分别使用了流的 XADD 命令、 XDEL 命令和 XLEN 命令;
  • 消息队列的两个获取方法 get_message() 和 get_by_range() 分别以两种形式调用了流的 XRANGE 命令;
  • 最后, 用于迭代消息的 iterate() 方法使用了 XREAD 命令对流进行迭代。

代码清单 10-1 使用 Redis 流实现的消息队列: /stream/message_queue.py

def reconstruct_message_list(message_list):
  """
  为了让多条消息能够以更结构化的方式返回给调用者,
  将 Redis 返回的多条消息从原来的格式:
  [(id1, {k1:v1, k2:v2, ...}), (id2, {k1:v1, k2:v2, ...}), ...]
  转换成以下格式:
  [{id1: {k1:v1, k2:v2, ...}}, {id2: {k1:v1, k2:v2, ...}}, ...]
  """  result = []
  for id, kvs in message_list:
    result.append({id: kvs})
  return result
def get_message_from_nested_list(lst):
  """
  从嵌套列表中取出消息本体。
  """
  return lst[0][1]
class MessageQueue:
  """
  使用 Redis 流实现的消息队列。
  """
  def __init__(self, client, stream_key):
    self.client = client
    self.stream = stream_key
  def add_message(self, key_value_pairs):
    """
    将给定的键值对存入到消息里面,并返回相应的消息 ID 。
    """
    return self.client.xadd(self.stream, key_value_pairs)
  def get_message(self, message_id):
    """
    根据给定的消息 ID 返回相应的消息,如果消息不存在则返回 None 。
    """
    reply = self.client.xrange(self.stream, message_id, message_id)
    if len(reply) == 1:
      return get_message_from_nested_list(reply)

  def remove_message(self, message_id):
    """
    根据给定的消息 ID 删除相应的消息,如果消息不存在则忽略该动作。
    """
    self.client.xdel(self.stream, message_id)

  def len(self):
    """
    返回消息队列的长度。
    """
    return self.client.xlen(self.stream)

  def get_by_range(self, start_id, end_id, max_item=10):
    """
    根据给定的 ID 区间范围返回队列中的消息。
    """
    reply = self.client.xrange(self.stream, start_id, end_id, max_item)
    return reconstruct_message_list(reply)

  def iterate(self, start_id=0, max_item=10):
    """
    对消息队列进行迭代,返回最多 N 条大于给定 ID 的消息。
    """
    reply = self.client.xread({self.stream: start_id}, max_item)
    if len(reply) == 0:
      return list()
    else:
      messages = get_message_from_nested_list(reply)
      return reconstruct_message_list(messages)

对于这个消息队列实现, 我们可以通过执行以下代码, 创建出它的实例:

>>> from redis import Redis
>>> from message_queue import MessageQueue
>>> client = Redis(decode_responses=True)
>>> mq = MessageQueue(client, "mq")

然后通过执行以下代码, 向队列里面添加十条消息:

>>> for i in range(10):
...  key = "key{0}".format(i)
...  value = "value{0}".format(i)
...  msg = {key:value}
...  mq.add_message(msg)
...
'1554113926280-0'
'1554113926280-1'
'1554113926281-0'
'1554113926281-1'
'1554113926281-2'
'1554113926281-3'
'1554113926281-4'
'1554113926281-5'
'1554113926281-6'
'1554113926282-0'

还可以根据 ID 获取指定的消息, 又或者使用 get_by_range() 方法同时获取多条消息:

>>> mq.get_message('1554113926280-0')
{'key0': 'value0'}
>>> mq.get_message('1554113926280-1')
{'key1': 'value1'}
>>> mq.get_by_range("-", "+", 3)
[{'1554113926280-0': {'key0': 'value0'}}, {'1554113926280-1': {'key1': 'value1'}}, {'1554113926281-0': {'key2': 'value2'}}]

又或者使用 iterate() 方法对消息队列进行迭代, 等等:

>>> mq.iterate(0, 3)
[{'1554113926280-0': {'key0': 'value0'}}, {'1554113926280-1': {'key1': 'value1'}}, {'1554113926281-0': {'key2': 'value2'}}]
>>> mq.iterate('1554113926281-0', 3)
[{'1554113926281-1': {'key3': 'value3'}}, {'1554113926281-2': {'key4': 'value4'}}, {'1554113926281-3': {'key5': 'value5'}}]

总结

以上所述是小编给大家介绍的使用 Redis 流实现消息队列的代码,希望对大家有所帮助,如果大家有任何疑问请给我留言,小编会及时回复大家的。在此也非常感谢大家对我们网站的支持!
如果你觉得本文对你有帮助,欢迎转载,烦请注明出处,谢谢!

时间: 2019-11-01

php+redis消息队列实现抢购功能

本文实例为大家分享了php+redis消息队列实现抢购的具体代码,供大家参考,具体内容如下 实现功能: 1. 基于redis队列,防止高并发的超卖 2. 基于mysql的事务加排它锁,防止高并发的超卖 基于redis队列工作流程: 1. 管理员根据goods表中的库存,创建redis商品库存队列 2. 客户端访问秒杀API 3. web服务器先从redis的商品库存队列中查询剩余库存重点内容 4. redis队列中有剩余,则在mysql中创建订单,去库存,抢购成功 5. redis队列中没有剩余

详解Redis用链表实现消息队列

前言 Redis链表经常会被用于消息队列的服务,以完成多程序之间的消息交换.个人认为redis消息队列有一个好处,就是可以实现分布式和共享,就和memcache作为mysql的缓存和mysql自带的缓存一样. 链表实现消息队列 Redis链表支持前后插入以及前后取出,所以如果往尾部插入元素,往头部取出元素,这就是一种消息队列,也可以说是消费者/生产者模型.可以利用lpush和rpop来实现.但是有一个问题,如果链表中没有数据,那么消费者将要在while循环中调用rpop,这样以来就浪费cpu资源

PHP基于Redis消息队列实现发布微博的方法

本文实例讲述了PHP基于Redis消息队列实现发布微博的方法.分享给大家供大家参考,具体如下: phpRedisAdmin :github地址  图形化管理界面 git clone [url]https://github.com/ErikDubbelboer/phpRedisAdmin.git[/url] cd phpRedisAdmin git clone [url]https://github.com/nrk/predis.git[/url] vendor 首先安装上述的Redis图形化管理

PHP使用php-resque库配合Redis实现MQ消息队列的教程

消息队列处理后台任务带来的问题 项目中经常会有后台运行任务的需求,比如发送邮件时,因为要连接邮件服务器,往往需要5-10秒甚至更长时间,如果能先给用户一个成功的提示信息,然后在后台慢慢处理发送邮件的操作,显然会有更好的用户体验. 为了实现类似的需求,Web项目中一般的实现方法是使用消息队列(Message Queue),比如MemcacheQ,RabbitMQ等等,都是很著名的产品. 消息队列说白了就是一个最简单的先进先出队列,队列的一个成员就是一段文本.正是因为消息队列实在太简单了,当拿着消息

phpredis提高消息队列的实时性方法(推荐)

数据库存贮都用list形式 要存2个队列 1个用作消息队列保存到数据 还有个 就是用来实时读取数据在redis $redis->lpush($queenkey, json_encode($array)); $redis->lpush($listkey, json_encode($array)); /*消息队列实例*/ public function insertinfo() { $infos = array('info1' => mt_rand(10,100), 'info2' =>

Java利用Redis实现消息队列的示例代码

本文介绍了Java利用Redis实现消息队列的示例代码,分享给大家,具体如下: 应用场景 为什么要用redis? 二进制存储.java序列化传输.IO连接数高.连接频繁 一.序列化 这里编写了一个java序列化的工具,主要是将对象转化为byte数组,和根据byte数组反序列化成java对象; 主要是用到了ByteArrayOutputStream和ByteArrayInputStream; 注意:每个需要序列化的对象都要实现Serializable接口; 其代码如下: package Utils

SpringBoot利用redis集成消息队列的方法

一.pom文件依赖 <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-redis</artifactId> </dependency> 二.创建消息接收者 变量.方法及构造函数进行标注,完成自动装配的工作. 通过 @Autowired的使用来消除 set ,get方法. @Autowired pub

php+redis实现消息队列功能示例

本文实例讲述了php+redis实现消息队列功能.分享给大家供大家参考,具体如下: 个人理解在项目中使用消息队列一般是有如下几个原因: 把瞬间服务器的请求处理换成异步处理,缓解服务器的压力 实现数据顺序排列获取 redis实现消息队列步骤如下: 1).redis函数rpush,lpop 2).建议定时任务入队列 3)创建定时任务出队列 文件:demo.php插入数据到redis队列 <?php $redis = new Redis(); $redis->connect('127.0.0.1',

python实现RabbitMQ的消息队列的示例代码

最近在研究redis做消息队列时,顺便看了一下RabbitMQ做消息队列的实现.以下是总结的RabbitMQ中三种exchange模式的实现,分别是fanout, direct和topic. base.py: import pika # 获取认证对象,参数是用户名.密码.远程连接时需要认证 credentials = pika.PlainCredentials("admin", "admin") # BlockingConnection(): 实例化连接对象 # C

C#调用RabbitMQ实现消息队列的示例代码

前言 我在刚接触使用中间件的时候,发现,中间件的使用并不是最难的,反而是中间件的下载,安装,配置才是最难的. 所以,这篇文章我们从头开始学习RabbitMq,真正的从头开始. 关于消息队列 其实消息队列没有那么神秘,我们这样想一下,用户访问网站,最终是要将数据以HTTP的协议的方式,通过网络传输到主机的某个端口上的. 那么,接收数据的方式是什么呢?自然是端口监听啦. 那消息队列是什么就很好解释了? 它就是端口监听,接到数据后,将数据排列起来. 那这件事,我们不用中间件能做吗? 当然能做啦,写个T

Spring boot 整合KAFKA消息队列的示例

这里使用 spring-kafka 依赖和 KafkaTemplate 对象来操作 Kafka 服务. 一.添加依赖和添加配置项 1.1.在 Pom 文件中添加依赖 <dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency> 1.2.添加配置项 spring: kafka: b

Java用数组实现循环队列的示例

复习了下数据结构,用Java的数组实现一下循环队列. 队列的类 //循环队列 class CirQueue{ private int QueueSize; private int front; private int rear; private int[] queueList ; public CirQueue(int QueueSize){ this.QueueSize = QueueSize; queueList = new int[QueueSize]; front = 0; rear =

java多线程消息队列的实现代码

本文介绍了java多线程消息队列的实现代码,分享给大家,希望对大家有帮助,顺便也自己留个笔记 1.定义一个队列缓存池: //static修饰的成员变量和成员方法独立于该类的任何对象.也就是说,它不依赖类特定的实例,被类的所有实例共享. private static List<Queue> queueCache = new LinkedList<Queue>(); 2.定义队列缓冲池最大消息数,如果达到该值,那么队列检入将等待检出低于该值时继续进行. private Integer

Python实现RabbitMQ6种消息模型的示例代码

RabbitMQ与Redis对比 ​ RabbitMQ是一种比较流行的消息中间件,之前我一直使用redis作为消息中间件,但是生产环境比较推荐RabbitMQ来替代Redis,所以我去查询了一些RabbitMQ的资料.相比于Redis,RabbitMQ优点很多,比如: 具有消息消费确认机制 队列,消息,都可以选择是否持久化,粒度更小.更灵活. 可以实现负载均衡 RabbitMQ应用场景 异步处理:比如用户注册时的确认邮件.短信等交由rabbitMQ进行异步处理 应用解耦:比如收发消息双方可以使用

Java实现8种排序算法的示例代码

冒泡排序 O(n2) 两个数比较大小,较大的数下沉,较小的数冒起来. public static void bubbleSort(int[] a) { //临时变量 int temp; //i是循环次数,也是冒泡的结果位置下标,5个数组循环5次 for (int i = 0; i < a.length; i++) { //从最后向前面两两对比,j是比较中下标大的值 for (int j = a.length - 1; j > i; j--) { //让小的数字排在前面 if (a[j] <