python - 多处理中的解密失败或记录不良

Question

我试图在填充 PostgreSQL 数据库的同时让所有 PC 内核同时工作，我已经编辑了代码以对我得到的内容产生可重现的错误

Traceback (most recent call last):
  File "test2.py", line 50, in <module>
    download_all_sites(sites)
  File "test2.py", line 36, in download_all_sites
    pool.map(download_site, sites)
  File "/usr/lib/python3.8/multiprocessing/pool.py", line 364, in map
    return self._map_async(func, iterable, mapstar, chunksize).get()
  File "/usr/lib/python3.8/multiprocessing/pool.py", line 771, in get
    raise self._value
psycopg2.OperationalError: SSL error: decryption failed or bad record mac

导致错误的完整代码

import requests
import multiprocessing
import time
import os
import psycopg2
session = None
conn = psycopg2.connect(user="user",
                        password="pass123",
                        host="127.0.0.1",
                        port="5432",
                        database="my_db")
cursor = conn.cursor()


def set_global_session():
    global session
    if not session:
        session = requests.Session()


def download_site(domain):
    url = "http://" + domain
    with session.get(url) as response:
        temp = response.text.lower()
        found = [i for i in keywords if i in temp]
        query = """INSERT INTO test (domain, keyword) VALUES (%s, %s)"""
        cursor.execute(query, (domain, found))


def download_all_sites(sites):
    with multiprocessing.Pool(processes=os.cpu_count(), initializer=set_global_session) as pool:
        pool.map(download_site, sites)


if __name__ == "__main__":
    sites = ['google.com'] * 10
    keywords = ['google', 'success']
    start_time = time.time()
    download_all_sites(sites)
    duration = time.time() - start_time
    conn.commit()
    print(f"Finished {len(sites)} in {duration} seconds")

Answer 1

为每个多进程创建一个新的 postgres 连接。 Libpq连接不应该与 fork 进程一起使用(多处理正在做什么)，它在 postgres docs 的第二个警告框中提到.

import requests
import multiprocessing
import time
import os
import psycopg2
session = None    

def set_global_session():
    global session
    if not session:
        session = requests.Session()


def download_site(domain):
    url = "http://" + domain
    with session.get(url) as response:
        #temp = response.text.lower()
        #found = [i for i in keywords if i in temp]
        #query = """INSERT INTO test (domain, keyword) VALUES (%s, %s)"""
        conn = psycopg2.connect(
            "dbname=mf port=5959 host=localhost user=mf_usr"
        )
        cursor = conn.cursor()
        query = """INSERT INTO mytable (name) VALUES (%s)"""
        cursor.execute(query, (domain, ))
        conn.commit()
        conn.close()


def download_all_sites(sites):
    with multiprocessing.Pool(
        processes=os.cpu_count(), initializer=set_global_session
    ) as pool:
        pool.map(download_site, sites)


if __name__ == "__main__":
    sites = ['google.com'] * 10
    keywords = ['google', 'success']
    start_time = time.time()
    download_all_sites(sites)
    duration = time.time() - start_time
    print(f"Finished {len(sites)} in {duration} seconds")

    # make sure it worked!
    conn = psycopg2.connect("dbname=mf port=5959 host=localhost user=mf_usr")
    cursor = conn.cursor()
    cursor.execute('select count(name) from mytable')
    print(cursor.fetchall())  # verify 10 downloads == 10 records in database

出去:

Finished 10 in 0.9922008514404297 seconds
[(10,)]