利用Redis监听有序队列（redis监听有序队列）

利用Redis监听有序队列

在现代软件架构中，消息队列是实现异步通信和事件驱动架构的常见方式。Redis是一种流行的内存数据库，同时也是一种有效的消息队列解决方案。在Redis中，有序集合可以实现基于优先级的消息队列。

有序集合可以对元素进行排序，并且支持添加、删除和查询操作。在Redis中，有序集合可以用于实现优先级队列。通过将元素放入有序集合中，并将分数作为优先级，我们就可以确保在读取元素时，按照优先级顺序读取。

在此基础上，我们可以使用Redis的pub/sub模式进行消息通知，以便在队列中有新元素时，立即处理该元素。使用pub/sub模式可以有效避免了轮询查询的操作。

下面是示例代码：

import redis
import threading
import time
class RedisPriorityQueue():

  def __init__(self, name, maxsize=0):
    self.rc = redis.Redis(host='localhost', port=6379, db=0)
    self.name = name
    self.maxsize = maxsize
  def qsize(self):
    return self.rc.zcard(self.name)
  def put(self, item, priority=0):
    self.rc.zadd(self.name, {item: priority})
    if self.maxsize > 0 and self.qsize() > self.maxsize:
      remove = self.rc.zrange(self.name, 0, 0, withscores=True)
      self.rc.zrem(self.name, remove[0][0])

  def get(self, block=True, timeout=None):
    while True:
      item = self.rc.zrange(self.name, 0, 0, withscores=True)
      if item:
        item = item[0][0]
        self.rc.zrem(self.name, item)
        return item
      else:
        if not block:
          return None
        if timeout is not None and timeout 
          return None
        time.sleep(0.1)

class RedisQueueListener():

  def __init__(self, name, callback):
    self.rc = redis.Redis(host='localhost', port=6379, db=0)
    self.name = name
    self.callback = callback
  def run(self):
    while True:
      item = self.rc.zrange(self.name, 0, 0, withscores=True)
      if item:
        item = item[0][0]
        self.rc.zrem(self.name, item)
        self.callback(item)
      time.sleep(0.1)
def callback(item):
  print('Received item: ' + item)
q = RedisPriorityQueue('myqueue')
l = RedisQueueListener('myqueue', callback)
t = threading.Thread(target=l.run)
t.daemon = True
t.start()

q.put('item1', 1)
q.put('item2', 2)
q.put('item3', 3)
q.put('item4', 4)
q.put('item5', 5)