我有一个应用程序，每 100 毫秒从传感器读取 float 据，将其附加到列表中，每 5 分钟从该列表中计算一些统计数据并插入到 MongoDB 数据库中。然后它会清理列表等等。

这些列表有很多(与传感器一样多)，我需要定期读取数据，所以我设置了 Celery worker。它工作得很好，但每个 Celery worker 都有自己特定的全局变量空间，因此在插入数据库期间列表具有不同的值，这取决于实际将数据插入数据库的 worker。

在工作人员之间共享数据并以某种方式锁定它以防止多个工作人员将其自己版本的传感器数据插入数据库的解决方案是什么？

我想到了 Redis 并将传感器数据直接附加到 Redis 字典，每 5 分钟从 Redis 读取数据、计算统计数据、清理 Redis 字典等等。

import celery
import my_data_reader
import my_stats_calculator
import my_mongo_manager

app = celery.Celery('tasks', broker='redis://localhost')

sensor_data = []

data_reader = my_data_reader.TemperatureReader(1)
mongo_writer = my_mongo_manager.DataWriter()
stats_calculator = my_stats_calculator.Calculator()


# Runs every 100 milliseconds
@app.task
def update_sensors():

    global sensor_data
    global data_reader

    sensor_data.append(data_reader.get_data())

# Runs every 5 seconds
@app.task
def insert_to_database():

    global sensor_data
    global mongo_writer
    global stats_calculator

    stats_dict = stats_calculator.calculate_stats(sensor_data)
    mongo_writer.insert_data(stats_dict)
    del sensor_data[:]

在使用 1 个进程(--concurrency=1 celery 标志)运行此代码后，它工作得非常好，但在实际项目中有超过 25 个传感器，我想以某种方式有效地执行这些操作。

有谁知道在工作人员之间共享这些对象的正确方法是什么？

最佳答案

我想出了如何使用 Redis 和其他一些东西来做到这一点。我展示了工作代码。如果有人知道更好的解决方案，请在此处发布。

首先，我为 Celery 任务编写了一个装饰器，以防止多个工作人员同时操作 Redis 数据。我做了一些研究，从 this site 找到了一个轻量级的

但是我看到还有其他选择可以使用第 3 方模块(例如 sherlock 或 celery_once)来实现。

import celery
import redis
import pymongo
from datetime import datetime as dt

app = celery.Celery('tasks', broker='redis://localhost')
redis_client = redis.Redis()

def only_one(function=None, key="", timeout=None):
    """Enforce only one celery task at a time."""

    def _dec(run_func):
        """Decorator."""

        def _caller(*args, **kwargs):
            """Caller."""
            ret_value = None
            have_lock = False
            lock = redis_client.lock(key, timeout=timeout)
            try:
                have_lock = lock.acquire(blocking=False)
                if have_lock:
                    ret_value = run_func(*args, **kwargs)
            finally:
                if have_lock:
                    lock.release()

            return ret_value

        return _caller

    return _dec(function) if function is not None else _dec

实现自定义任务 - 运行方法现在由我们的 Redis 锁修饰

class SensorTask(app.Task):
    """A task."""

    @only_one(key='SensorTask', timeout=60 * 5)
    def run(self, **kwargs):
        # Append some data to redis list
        redis_client.lpush('Sensor1', 1.50)


class DatabaseTask(app.Task):
    """A task."""

    # Database connection will stay the same in each process
    # See https://docs.celeryproject.org/en/latest/userguide/tasks.html
    _mongo_client = None

    @property
    def mongo_client(self):
        if self._mongo_client is None:
            self._mongo_client = pymongo.MongoClient()
        return self._mongo_client

    @only_one(key='DatabaseTask', timeout=60 * 5)
    def run(self, **kwargs):

        # Read current list of sensor values from Redis
        current_sensor_values = redis_client.lrange('Sensor1', 0, -1)

        # Convert Redis list to python float list
        # map compares to list comprehension is a bit faster in my case
        # values = [float(i) for i in current_sensor_values]
        values = list(map(float, current_sensor_values))

        # Example Mongo document to insert after 5 minutes of collecting data
        mongo_document = {
                'Timestamp': dt.now(),
                'first': values[0],
                'last': values[-1],
                'max' : max(values),
                'min' : min(values)
                }

        # Insert document to Mongo database and clean the Redis list
        self.mongo_client['Sensors']['Sensor1'].insert_one(mongo_document)
        redis_client.delete('Sensor1')

最后一步是将我们的任务注册到 Celery 空间:

update_sensor = app.register_task(SensorTask())
update_database = app.register_task(DatabaseTask()) 

现在它在多个工作器上工作得很好。要运行任务，您需要使用创建的别名调用它 - 在我们的例子中是 update_sensor.delay() 和 update_database.delay()

关于python - 在 Celery - Python 中的进程(工作人员)之间共享数据的最佳解决方案是什么？，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/57275700/