Python中存储回调函数数据的方法——基于roslibpy的ROS主题订阅场景

我经常碰到这类异步回调场景下的数据存储问题，roslibpy的subscribe设计确实是典型的事件驱动模式——只通过回调传递数据，不会直接返回结果。不过有几种通用方案可以解决你的需求，我们一步步来看：

一、内存存储：脚本运行期间实时访问（最常用） #

如果只是需要在脚本运行过程中随时访问这些数据，内存存储是最高效的选择，有两种优雅的实现方式：

1. 使用全局变量（快速上手） #

你可以定义一个全局容器（比如列表或字典），在回调函数里把收到的消息添加进去：

import roslibpy

client = roslibpy.Ros(host='localhost', port=9090)
client.run()

# 全局容器存储pose数据
pose_data = []

def myfunc(msg):
    print(msg)
    # 将消息添加到全局列表
    pose_data.append(msg)

listener = roslibpy.Topic(client, '/turtle1/pose', 'turtlesim/Pose')
listener.subscribe(myfunc)

try:
    while True:
        # 这里可以随时访问pose_data，比如打印最新的一条
        if pose_data:
            print(f"最新Pose数据: {pose_data[-1]}")
        # 加个小延迟避免占用过多CPU
        roslibpy.sleep(0.1)
except KeyboardInterrupt:
    client.terminate()

⚠️ 注意：如果你的脚本涉及多线程操作（比如在其他线程读写pose_data），最好给容器加个线程锁（比如threading.Lock），避免数据竞争。

2. 使用类封装（更优雅的工程化方式） #

全局变量在复杂脚本里容易混乱，用类封装可以把数据和订阅逻辑放在一起：

import roslibpy

class PoseSubscriber:
    def __init__(self, client):
        self.pose_data = []
        self.topic = roslibpy.Topic(client, '/turtle1/pose', 'turtlesim/Pose')
        # 订阅时绑定类的回调方法
        self.topic.subscribe(self.handle_pose)
    
    def handle_pose(self, msg):
        print(msg)
        self.pose_data.append(msg)

client = roslibpy.Ros(host='localhost', port=9090)
client.run()

pose_listener = PoseSubscriber(client)

try:
    while True:
        if pose_listener.pose_data:
            print(f"当前已存储{len(pose_listener.pose_data)}条Pose数据")
        roslibpy.sleep(0.1)
except KeyboardInterrupt:
    client.terminate()

这种方式更易于维护，尤其是当你需要订阅多个主题或者添加更多数据处理逻辑时。

二、持久化存储：长期保存或跨脚本访问 #

如果需要长期保存数据，或者脚本重启后还要使用这些数据，才需要考虑写入外部源（比如.txt、JSON文件、数据库）。但要注意频繁IO操作会有性能损耗，可以参考以下优化方式：

1. 写入文本文件（简单直接） #

可以在回调里把消息转成字符串写入文件，或者攒一批数据再批量写入：

import roslibpy
import json

client = roslibpy.Ros(host='localhost', port=9090)
client.run()

# 攒10条数据再写入一次，减少IO次数
batch_size = 10
temp_buffer = []

def myfunc(msg):
    print(msg)
    temp_buffer.append(msg)
    # 达到批量大小就写入文件
    if len(temp_buffer) >= batch_size:
        with open('pose_data.txt', 'a', encoding='utf-8') as f:
            for data in temp_buffer:
                # 转成JSON字符串方便后续读取
                f.write(json.dumps(data) + '\n')
        temp_buffer.clear()

listener = roslibpy.Topic(client, '/turtle1/pose', 'turtlesim/Pose')
listener.subscribe(myfunc)

try:
    while True:
        pass
except KeyboardInterrupt:
    # 退出前把剩余的缓冲数据写入
    if temp_buffer:
        with open('pose_data.txt', 'a', encoding='utf-8') as f:
            for data in temp_buffer:
                f.write(json.dumps(data) + '\n')
    client.terminate()

2. 什么时候适合用持久化？ #

• 需要长期保存数据用于后续分析
• 脚本重启后需要恢复之前的数据
• 其他程序需要读取这些数据

如果只是脚本运行期间需要访问数据，内存存储完全足够，没必要额外增加IO开销。

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