时间戳

首页 > 解决方案 > 无法从 Redis 订阅中获取数据？

python - 无法从 Redis 订阅中获取数据？

问题描述

我正在尝试通过在我的客户端应用程序上使用订阅从 redis 通道获取数据。为此，我将 python 与 asyncio 和 aioredis 一起使用。

我想使用我的订阅来更新我的主应用程序的一个变量，当这个变量在服务器上发生更改时，我无法设法将从订阅接收到的数据传递给我的主线程。

根据 aioredis网站，我通过以下方式实现了我的订阅：

sub = await aioredis.create_redis(
     'redis://localhost')

ch1 = await sub.subscribe('channel:1')
assert isinstance(ch1, aioredis.Channel)

async def async_reader(channel, globarVar):
    while await channel.wait_message():
        msg = await channel.get(encoding='utf-8')
        # ... process message ...
        globarVar = float(msg)
        print("message in {}: {}".format(channel.name, msg))

tsk1 = asyncio.ensure_future(async_reader(ch1, upToDateValue))

但我无法更新全局变量，我猜 python 只将当前值作为参数传递（这是我期望的，但想确定）。

是否有任何可行的选择从订阅中获取数据？或者传递对我可以使用的共享变量或队列的引用？

标签： pythonrediscallbackpython-asyncio

解决方案

你应该重新设计你的代码，这样你就不需要全局变量了。您的所有处理都应在收到消息时进行。但是，要修改全局变量，您需要在函数中使用 global 关键字声明它。你不传递全局变量——你只是使用它们。

子：

import aioredis
import asyncio
import json

gvar = 2

# Do everything you need here or call another function
# based on the message.  Don't use a global variable.
async def process_message(msg):
  global gvar
  gvar = msg

async def async_reader(channel):
  while await channel.wait_message():
    j = await channel.get(encoding='utf-8')
    msg = json.loads(j)
    if msg == "stop":
      break
    print(gvar)
    await process_message(msg)
    print(gvar)

async def run(loop):
  sub = await aioredis.create_redis('redis://localhost')
  res = await sub.subscribe('channel:1')
  ch1 = res[0]
  assert isinstance(ch1, aioredis.Channel)

  await async_reader(ch1)

  await sub.unsubscribe('channel:1')
  sub.close()

loop = asyncio.get_event_loop()
loop.run_until_complete( run(loop) )
loop.close()

出版商：

import asyncio
import aioredis

async def main():
    pub = await aioredis.create_redis('redis://localhost')

    res = await pub.publish_json('channel:1', ["Hello", "world"])
    await asyncio.sleep(1)
    res = await pub.publish_json('channel:1', "stop")

    pub.close()


if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())

推荐阅读