多处理和连接池错误:无法pickle'psycopg2.extensions.connection'对象

问题描述 投票:0回答:1

我试图将一个使用 Postgres 中的连接池的数据库对象传递给另一个类对象,但我不断收到错误。

if __name__ == "__main__":
cores = calculate_number_of_cores()
pretty_print(f"Number of cores ==> {cores}")

db_manager = DatabaseManager()
with ProcessPoolExecutor(cores) as executor:
    while True:
        if LOCAL_RUN:
            pretty_print("ALERT: Doing a local run of the automation with limited capabilities.")

        list_of_clients = db_manager.get_clients()
        print(list_of_clients)
        random.shuffle(list_of_clients)

        list_of_not_attempted_clients_domains = db_manager.get_not_attempted_domains_that_matches_random_client_tags()
        group_of_clients_handlers = {}
        pretty_print(list_of_not_attempted_clients_domains)

        # no matches
        if not list_of_not_attempted_clients_domains:
            sleep = 60 * 10
            pretty_print(f'No matches found. Sleeping for {sleep}s')
            time.sleep(sleep)
            continue

        for client in list_of_clients:
            client_id = client[0]
            client_name = client[1]
            group_of_clients_handlers[client_id] = [ClientsHandler(db_manager), client_name]

        try:
            list(executor.map(
                partial(run, group_of_clients_handlers),
                list_of_not_attempted_clients_domains
            ))
            # for not_attempted_clients_domains in list_of_not_attempted_clients_domains:
            #     futures = executor.submit(run(group_of_clients_handlers, not_attempted_clients_domains))
            #     for future in concurrent.futures.as_completed(futures):
            #         pretty_print("Loading futures....")
            #         print(future.result())

            time.sleep(60)
            pretty_print("sleeping for 60secs")
        except Exception as err:
            pretty_print(err)

我正在Python中使用多处理模块。我的目标是为每个数据库查询准备好连接,而不是每次执行或进程都必须连接到数据库。

我必须将

db_manager
传递给
run
函数,因为我不想在那里创建新连接。但正因为如此,我收到错误:无法pickle 'psycopg2.extensions.connection' 对象。

我什至不知道这到底意味着什么。我唯一知道的是,当我手动将

run
函数提交给执行器时,代码可以正常工作。这就是注释掉的代码。

我不知道还有什么其他方法可以解决这个问题。我有点知道错误来自哪里,但如何解决它,使其不干扰我的设置真是太痛苦了。

python postgresql parallel-processing multiprocessing connection-pooling
1个回答
0
投票

您收到的错误意味着

connection
对象无法使用pickle序列化。当您使用多处理来生成子进程时,某些平台上会发生对象的酸洗,您可以在这个答案中阅读更多相关信息。

酸洗并不是您的主要问题。即使您在生成子进程时避免使用 pickle,也不应该尝试与其他进程共享 psycopg 连接。根据 psycopg2 文档,您不应该在进程之间共享连接。

如果您需要预初始化连接池并且数据库操作非常耗时,您应该尝试使用

threading
而不是
multiprocessing
,因为根据
psycopg2
文档(上面的链接)连接是线程安全的。使用线程,您应该能够根据需要实现您的程序,在主线程中使用预初始化的连接池,并使用该池将查询委托给其他线程。

© www.soinside.com 2019 - 2024. All rights reserved.