标签: py-redis

我正在使用 redis-py 2.10.6 和 redis 4.0.11。

我的应用程序使用 redis 作为数据库和 pubsub。当我关闭时，我经常会遇到挂起或崩溃的情况。后者通常会抱怨处理 pubsub 回调时发生的错误文件描述符或文件上的 I/O 错误（我不使用任何错误），所以我猜测根本问题是相同的：不知何故我不'无法正确断开连接，并且我的 redis.Redis 对象使用的池处于活动状态。

前一种错误的输出示例（在 _read_from_socket 期间）：

redis.exceptions.ConnectionError：从套接字读取时出错：（9，'错误文件描述符'）

其他时候，堆栈跟踪清楚地显示 redis/connection.py -> redis/client.py -> threading.py，这证明 redis 没有杀死它使用的线程。

当我启动应用程序时，我运行：

self.redis = redis.Redis(host=XXXX, port=XXXX)
self.pubsub = self.redis.pubsub()
subscriptions = {'chan1': self.cb1, 'chan2': self.cb2}  # cb1 and cb2 are functions
self.pubsub.subscribe(**subscriptions)
self.pubsub_thread = self.pubsub.run_in_thread(sleep_time=1)

当我想退出应用程序时，我在main中执行的最后一条指令是调用我的 redis using 类中的函数，其实现是：

self.pubsub.close()
self.pubsub_thread.stop()
self.redis.connection_pool.disconnect()

我的理解是，理论上我什至不需要执行任何这些“关闭”调用，但是，无论有或没有它们，我仍然无法保证彻底关闭。

我的问题是，我该如何保证干净关闭？

使用

dd = {'ID': ['H576','H577','H578','H600', 'H700'],
      'CD': ['AAAAAAA', 'BBBBB', 'CCCCCC','DDDDDD', 'EEEEEEE']}
df = pd.DataFrame(dd)

在Pandas 0.25之前，这在下面起作用。

set:  redisConn.set("key", df.to_msgpack(compress='zlib'))
get:  pd.read_msgpack(redisConn.get("key"))

现在，已弃用警告。

FutureWarning: to_msgpack is deprecated and will be removed in a future version.
It is recommended to use pyarrow for on-the-wire transmission of pandas objects.

The read_msgpack is deprecated and will be removed in a future version.
It is recommended to use pyarrow for on-the-wire transmission of pandas objects.

罂粟如何运作？而且，我如何使pyarrow对象进出Redis。

参考： 如何在Redis中设置pandas.DataFrame或从Redis获取pandas.DataFrame？

我目前正在尝试对股价烛台进行简单的实现。假设我们有一只名为 XYZ 的股票。该股票接收一系列价格（没有特定频率），其（例如）如下所示：XYZ：[10.2, 10.7, 12, 11 ....]。

目标是记录每一分钟的一些指标，以反映该股票的状态。烛台具有收盘价（一分钟内的最后已知价格）、最高价（一分钟内的最高价格）等指标。

我认为可以实现此目的的一种方法是使用Redis TimeSeries。我考虑这个解决方案是因为我可以在价格流上创建规则，并且每 60 秒它会将一些聚合（例如：最大、最小、第一个......等）刷新到目标存储桶。

我当前使用 Redis TimeSeries （在Python中）作为每个股票价格的蜡烛图的实现看起来像这样（再次使用股票 XYZ 作为示例），并且为了简单起见没有标签：

from redistimeseries.client import Client
r = Client()
r.flushdb()

# Create source & destination buckets
r.create('XYZ_PRICES')  # source
r.create(closing_price)
r.create(highest_price)
# Create rules to send data from src -> closing_price & highest_price buckets
r.createrule(src, 'closing_price', 'last', bucket_size_msec=60000)
r.createrule(src, 'highest_price', 'max', bucket_size_msec=60000)

我的问题是：

1. 有没有一种方法可以在一条规则中发送多个聚合（例如 max、last...等），而不是为每个股票创建多个源和目标存储桶？
2. Redis TimeSeries 是否适合此任务？或者使用其他解决方案（例如 Redis 流）会更容易吗？