Node.js与Python结合Redis实现HSET变更监听的最优方案咨询

我用Node.js编写了一个基础Web API，将对象以HSET的形式写入Docker容器中的Redis缓存。同一虚拟机（VM）上运行的Python脚本需要监听Redis缓存，在有新HSET创建或HSET字段发生变更时执行相关代码。

我了解到Redis Pub/Sub，编写了两个测试用的Python脚本（订阅和发布），计划将发布端改写为Node.js，仅发布HSET的键（如id:3），然后Python订阅端收到消息后查询对应的HSET并处理，但觉得这种方式不妥，且Redis Watch不支持HSET，想寻找更优的实现方案。

测试用订阅脚本：

import redis
import json

print("Redis Subscriber")

redis_conn = redis.Redis(
    host='localhost',
    port=6379, 
    password='xxx', 
    charset="utf-8", 
    decode_responses=True)

def sub():
    pubsub = redis_conn.pubsub()
    pubsub.subscribe("broadcast")
    for message in pubsub.listen():
        if message.get("type") == "message":
            data = json.loads(message.get("data"))
            print(data)

if __name__ == "__main__":
    sub()

测试用发布脚本：

import redis
import json

print("Redis Publisher")

redis_conn = redis.Redis(
    host='localhost',
    port=6379, 
    password='xxx', 
    charset="utf-8", 
    decode_responses=True)

def pub():
    data = {
        "message": "id:3"
    }
    redis_conn.publish("broadcast", json.dumps(data))

if __name__ == "__main__":
    pub()

方案1：优化Pub/Sub模式（可控性强） #

你的初始思路其实具备可行性，只需补充可靠性措施即可落地：

• Node.js端改造：在执行HSET/HSETNX操作后，立即将HSET的键名（或更详细的变更信息）发布到指定频道。如果是更新操作，可附带变更字段列表，减少Python端的查询量。
  示例Node.js代码：

  const redis = require('redis');
  const client = redis.createClient({
    host: 'localhost',
    port: 6379,
    password: 'xxx'
  });
  
  // 写入HSET并发布通知
  async function setHsetAndNotify(key, fields) {
    // 添加更新时间戳，用于后续去重校验
    const payload = { ...fields, last_updated: new Date().toISOString() };
    await client.hSet(key, payload);
    // 仅发布键名，或发布包含变更字段的详细信息
    await client.publish('hset_updates', key);
  }
  

• Python订阅端优化：收到键名后拉取完整HSET数据，通过本地缓存记录已处理的键和最新更新时间，避免重复处理。
  改进后的Python订阅脚本：

  import redis
  
  redis_conn = redis.Redis(
      host='localhost',
      port=6379, 
      password='xxx', 
      decode_responses=True)
  
  # 本地缓存：记录已处理的键及其最后更新时间
  processed_records = {}
  
  def sub():
      pubsub = redis_conn.pubsub()
      pubsub.subscribe("hset_updates")
      for message in pubsub.listen():
          if message.get("type") == "message":
              key = message.get("data")
              hset_data = redis_conn.hgetall(key)
              last_updated = hset_data.get('last_updated', '')
              
              # 校验是否为未处理的最新版本
              if processed_records.get(key) != last_updated:
                  print(f"Processing HSET: {key}, Data: {hset_data}")
                  processed_records[key] = last_updated
                  # 此处执行你的业务逻辑
  
  if __name__ == "__main__":
      sub()
  

方案2：使用Redis键空间通知（无需业务代码侵入） #

如果不想手动维护Pub/Sub逻辑，可以使用Redis内置的键空间通知，它能自动监听键的创建、修改、删除等事件。

步骤1：开启Redis键空间通知 #

修改Redis配置文件redis.conf，或在运行时执行命令开启：

# 开启哈希表相关事件（HSET/HDEL等）+ 键通用事件（创建/删除）
CONFIG SET notify-keyspace-events Khg

• K：监听键空间事件（对应频道__keyspace@0__:*）
• h：监听哈希表相关命令
• g：监听键的通用操作（如创建、删除）

步骤2：Python端订阅事件 #

直接订阅对应事件频道，收到通知后查询HSET数据：

import redis

redis_conn = redis.Redis(
    host='localhost',
    port=6379, 
    password='xxx', 
    decode_responses=True)

def listen_hset_events():
    pubsub = redis_conn.pubsub()
    # 订阅所有键空间事件
    pubsub.psubscribe('__keyspace@0__:*')
    for message in pubsub.listen():
        if message.get("type") == "pmessage":
            channel = message.get("channel")
            key = channel.split(':')[-1]
            event = message.get("data")
            # 仅处理哈希表相关操作
            if event in ['hset', 'hsetnx', 'hmset', 'hdel'] or (event == 'set' and redis_conn.type(key) == 'hash'):
                print(f"Detected {event} on HSET key: {key}")
                hset_data = redis_conn.hgetall(key)
                # 此处执行你的业务逻辑

if __name__ == "__main__":
    listen_hset_events()

方案对比 #

关于Redis Watch的说明 #

Redis Watch是事务中的乐观锁机制，仅用于在事务执行前检测键是否被修改，完全不适合持续监听键变更的场景，无需考虑。

内容的提问来源于stack exchange，提问作者user1624184