使用Multiprocessing与PyMongo时遭遇[Errno 111]连接拒绝问题求助

我来帮你搞定这个多进程和PyMongo结合的连接问题！你遇到的[Errno 111] Connection refused本质原因是PyMongo的MongoClient对象不能跨进程共享——你在主进程创建的myClient被所有子进程复用，但进程间的资源隔离会导致这个客户端连接在子进程中失效，最终触发连接超时。

核心解决方案：每个子进程独立创建MongoClient #

不要在主进程初始化MongoClient，而是把客户端的创建逻辑移到每个子进程执行的createCollectionPDB函数内部。这样每个子进程都有自己独立的MongoDB连接，避免进程间的资源冲突。

修改后的createCollectionPDB函数： #

def createCollectionPDB(fP):
    # 每个子进程独立创建MongoClient
    myClient = MongoClient('mongodb://127.0.0.1:27017/')
    lineName = ""
    lineFASTA = ""
    colName = ""
    PDBName = ""
    chainIDName = ""
    typeOfMol = ""
    molLen = ""
    proteinName = ""
    for i in fP:
        if ">" in i:
            lineName = i.strip()
            colName = lineName.split(" ")[0].strip()[1:]
            PDBName = lineName.split("_")[0].strip()
            chainIDName = colName.split("_")[-1].strip()
            typeOfMol = lineName.split(" ")[1].strip().split(":")[1].strip()
            molLen = lineName.split(" ")[2].strip().split(":")[-1].strip()
            proteinName = lineName.split(" ")[-1].strip()
        else:
            lineFASTA = i.strip()
            dict2Write={"PDB_ID" : PDBName, "Chain_ID" : chainIDName, "Molecule Type" : typeOfMol, "Length" : molLen, "Protein_Name" : proteinName, "FASTA" : lineFASTA}
            myNewDB = myClient["MyPrjPrj_PDBs"]
            newCol = myNewDB[colName]
            x = newCol.insert_one(dict2Write)
            print("PDB", x.inserted_id)
    # 子进程结束前关闭客户端连接
    myClient.close()

优化方案：用进程池初始化函数复用子进程的MongoClient #

如果担心每次函数调用都创建客户端影响性能，可以用进程池的initializer参数，为每个子进程初始化一次MongoClient，这样每个子进程的所有任务都复用同一个客户端：

from multiprocessing import *
from pymongo import MongoClient

# 子进程全局变量，存储客户端
worker_client = None

def init_worker():
    """进程池初始化函数，每个子进程启动时执行"""
    global worker_client
    worker_client = MongoClient('mongodb://127.0.0.1:27017/')

def listFile(fileName):
    # 保持原逻辑不变
    fOpen = open(fileName)
    listFile = fOpen.readlines()
    arrOfArrs = []
    tmp1 = []
    for i in listFile:
        if i.startswith(">"):
            if len(tmp1) > 1:
                arrOfArrs.append(tmp1)
            tmp1 = []
            tmp1.append(i.strip())
        else:
            tmp1.append(i.strip())
    return arrOfArrs

def createCollectionPDB(fP):
    global worker_client
    lineName = ""
    lineFASTA = ""
    colName = ""
    PDBName = ""
    chainIDName = ""
    typeOfMol = ""
    molLen = ""
    proteinName = ""
    for i in fP:
        if ">" in i:
            lineName = i.strip()
            colName = lineName.split(" ")[0].strip()[1:]
            PDBName = lineName.split("_")[0].strip()
            chainIDName = colName.split("_")[-1].strip()
            typeOfMol = lineName.split(" ")[1].strip().split(":")[1].strip()
            molLen = lineName.split(" ")[2].strip().split(":")[-1].strip()
            proteinName = lineName.split(" ")[-1].strip()
        else:
            lineFASTA = i.strip()
            dict2Write={"PDB_ID" : PDBName, "Chain_ID" : chainIDName, "Molecule Type" : typeOfMol, "Length" : molLen, "Protein_Name" : proteinName, "FASTA" : lineFASTA}
            myNewDB = worker_client["MyPrjPrj_PDBs"]
            newCol = myNewDB[colName]
            x = newCol.insert_one(dict2Write)
            print("PDB", x.inserted_id)

if __name__ == "__main__":
    x = cpu_count()
    print(x)
    f1 = listFile("datum/pdb_seqres.txt")
    # 初始化进程池时指定初始化函数
    myPool = Pool(processes=x, initializer=init_worker)
    myPool.map(createCollectionPDB, f1)
    myPool.close()
    myPool.join()

额外排查要点 #

• 确认MongoDB服务状态：
  运行命令检查服务是否正常启动：

  sudo systemctl status mongod
  

  如果未启动，执行：

  sudo systemctl start mongod
  

• 验证MongoDB监听地址：
  确保MongoDB确实在监听127.0.0.1:27017：

  netstat -tulpn | grep 27017
  

• 检查本地防火墙：
  确保Ubuntu的ufw没有拦截本地MongoDB连接：

  sudo ufw status
  

  如果有必要，可以临时关闭防火墙测试：

  sudo ufw disable
  

内容的提问来源于stack exchange，提问作者user2919828