我将如何在 python 中创建一个 (async/threaded/task) 背景队列？

Question

对于 C# 应用程序，我使用了一个后台队列，我可以在其中排队 'actions'。我希望在 Python 中做同样的事情。

后台队列应该 'enqueue' 一个 'action' 包含对函数的调用（有或没有变量）并且应该在主程序继续执行其自己的功能时继续执行任务。

我已经尝试过使用 rq，但似乎不起作用。我很想听听一些建议！

编辑：这段代码是关于：

class DatabaseHandler:
def __init__(self):
    try:
        self.cnx = mysql.connector.connect(user='root', password='', host='127.0.0.1', database='mydb')
        self.cnx.autocommit = True
        self.loop = asyncio.get_event_loop()
    except mysql.connector.Error as err:
        if err.errno == errorcode.ER_ACCESS_DENIED_ERROR:
            print("Something is wrong with your user name or password")
        elif err.errno == errorcode.ER_BAD_DB_ERROR:
            print("Database does not exist")
        else:
            print(err)
    self.get_new_entries(30.0)

    async def get_new_entries(self, delay):
        start_time = t.time()
        while True:
            current_time = datetime.datetime.now() - datetime.timedelta(seconds=delay)
            current_time = current_time.strftime("%Y-%m-%d %H:%M:%S")
            data = current_time
            print(current_time)
            await self.select_latest_entries(data)
            print("###################")
            t.sleep(delay - ((t.time() - start_time) % delay))

    async def select_latest_entries(self, input_data):
        query = """SELECT FILE_NAME FROM `added_files` WHERE CREATION_TIME > %s"""
        cursor = self.cnx.cursor()
        await cursor.execute(query, (input_data,))
        async for file_name in cursor.fetchall():
            file_name_string = ''.join(file_name)
            self.loop.call_soon(None, self.handle_new_file_names, file_name_string)
        cursor.close()

    def handle_new_file_names(self, filename):
        # self.loop.run_in_executor(None, NF.create_new_npy_files, filename)
        # self.loop.run_in_executor(None, self.update_entry, filename)
        create_new_npy_files(filename)
        self.update_entry(filename)

    def update_entry(self, filename):
        print(filename)
        query = """UPDATE `added_files` SET NPY_CREATED_AT=NOW(), DELETED=1 WHERE FILE_NAME=%s"""
        update_cursor = self.cnx.cursor()
        self.cnx.commit()
        update_cursor.execute(query, (filename,))
        update_cursor.close()

create_new_npy_files(filename) 是来自静态 class 的静态方法，如果这有意义的话。这是一个非常耗时的函数（1-2 秒）

Answer 1

如果要执行的动作很短non-blocking，可以使用call_soon:

loop = asyncio.get_event_loop()
loop.call_soon(action, args...)

如果操作可能需要更长的时间或可能会阻塞，请使用 run_in_executor 将它们提交到线程池：

loop = asyncio.get_event_loop()
future = loop.run_in_executor(None, action, args...)
# you can await the future, access its result once ready, etc.

请注意，以上两个片段都假设您已经在程序中使用了 asyncio，基于 python-asyncio 标记。这意味着您的 select_statement 将如下所示：

async def select_statement():
    loop = asyncio.get_event_loop()
    while True:
        # requires an async-aware db module
        await cursor.execute(query, (input_data,))
        async for file_name in cursor.fetchall():
            loop.call_soon(self.handle_new_file_names, file_name_string))
            # or loop.run_in_executor(...)

我将如何在 python 中创建一个 (async/threaded/task) 背景队列？

How would I go and create an (async/threaded/task) backgroundqueue in python?

python

redis

python-3.x

python-asyncio