Celery多Worker进程间数据共享与同步问题解决方案咨询

这是Celery多实例部署中非常常见的状态共享难题——每个Worker都是独立进程，全局变量完全隔离，所以你之前单Worker正常、多Worker数据分散的情况完全符合预期。你提到的用Redis存储数据的思路方向是对的，不过还要补上分布式锁和多传感器适配的结构设计，才能保证数据一致性和处理效率，下面详细说下具体实现：

一、用Redis作为统一的传感器数据存储层 #

Celery Worker的进程隔离性决定了本地全局变量没法共享，必须用外部存储来做数据中转。Redis的列表（List）结构天然适合你的场景：

• 每个传感器对应一个独立的Redis键，比如用sensor:{sensor_id}:raw_data作为键名，这样25+传感器的数据不会互相干扰。
• Redis的RPUSH（或LPUSH）操作是原子性的，多个Worker同时往同一个传感器的列表追加数据也不会出现数据丢失或错乱。

二、分布式锁避免统计计算时的并发冲突 #

当执行5分钟一次的统计任务时，可能存在多个Worker同时触发同一传感器的统计逻辑，这时候会出现重复读取、重复计算或者清空不彻底的问题。所以必须给每个传感器的统计操作加分布式锁：

• 用Redis的锁机制（推荐用redis-py自带的Lock类），确保同一时间只有一个Worker能处理某传感器的统计、入库和清空操作。
• 锁的粒度要控制在单个传感器级别，不要用全局锁，否则会影响多传感器的并行处理效率。

三、重构任务代码适配多传感器场景 #

针对25+传感器的情况，要把任务改成可复用的参数化形式，避免为每个传感器写重复代码：

import celery
import redis
import my_data_reader
import my_stats_calculator
import my_mongo_manager

# 初始化Celery
app = celery.Celery('tasks', broker='redis://localhost')
# 初始化Redis客户端（用于数据存储和锁）
redis_client = redis.Redis(host='localhost', port=6379, db=0)
# 初始化Mongo和统计器（这些是无状态的，可以全局初始化）
mongo_writer = my_mongo_manager.DataWriter()
stats_calculator = my_stats_calculator.Calculator()

# 每100ms执行一次，读取指定传感器数据并写入Redis
@app.task
def update_sensor(sensor_id):
    # 根据sensor_id初始化对应的数据读取器（如果读取器是无状态的，也可以缓存起来）
    data_reader = my_data_reader.SensorReader(sensor_id)
    raw_data = data_reader.get_data()
    # 原子追加到Redis列表
    redis_client.rpush(f"sensor:{sensor_id}:raw_data", raw_data)

# 每5分钟执行一次，处理指定传感器的统计与入库
@app.task
def process_sensor_stats(sensor_id):
    redis_key = f"sensor:{sensor_id}:raw_data"
    lock_key = f"lock:sensor:{sensor_id}:processing"
    
    # 获取分布式锁，超时时间设为统计+入库的预估时间，避免死锁
    with redis_client.lock(lock_key, timeout=30):
        # 读取所有原始数据
        raw_data_list = redis_client.lrange(redis_key, 0, -1)
        # 注意：Redis存的是字节，需要转成浮点型
        raw_data = [float(item) for item in raw_data_list]
        
        if raw_data:
            # 计算统计量
            stats_dict = stats_calculator.calculate_stats(raw_data)
            # 写入MongoDB，建议给stats_dict加上sensor_id字段方便查询
            stats_dict["sensor_id"] = sensor_id
            mongo_writer.insert_data(stats_dict)
            
            # 清空Redis中的原始数据（原子操作）
            redis_client.delete(redis_key)

四、任务调度与性能优化建议 #

1. Celery Beat配置：在celerybeat-schedule中配置每个传感器的update_sensor和process_sensor_stats任务，比如：

   [schedule]
   update_sensor_1 = {task: 'tasks.update_sensor', schedule: 0.1, args: [1]}
   update_sensor_2 = {task: 'tasks.update_sensor', schedule: 0.1, args: [2]}
   # ... 其他传感器的update任务
   process_sensor_1 = {task: 'tasks.process_sensor_stats', schedule: 300, args: [1]}
   process_sensor_2 = {task: 'tasks.process_sensor_stats', schedule: 300, args: [2]}
   # ... 其他传感器的process任务
   

2. Redis批量操作：如果传感器数据量很大，LRANGE可以分批次读取，但你的场景是5分钟清空一次，数据量不会特别大，一次性读取没问题。
3. 读取器复用：如果SensorReader初始化成本高，可以用Redis或者本地缓存（比如lru_cache）来复用实例，减少资源消耗。
4. 监控与告警：给Redis和Celery加监控，比如监控Redis内存使用、任务执行失败率，避免数据堆积或任务丢失。

为什么不用Celery的状态存储？ #

Celery自带的result_backend主要用于存储任务结果，不适合用来做实时的数据流存储，而且性能和灵活性远不如Redis，所以用Redis作为中转是最优选择。

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