python - 我将如何在 python 中创建一个(异步/线程/任务)后台队列？

Question

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

后台队列应该“入队”一个“ 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) 是来自静态类的静态方法。这是一个非常耗时的函数(1-2 秒)

Answer 1

如果要执行的 Action 很短且非阻塞，您可以使用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.

请注意，以上两个片段都假定您已经在程序中使用了基于 python-asyncio 标记的 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(...)