使用多处理和 PySftp 并行下载

Question

我正在尝试创建一个代码来使用 pysftp 和多处理库下载 N 个相同类型的文件。我进行了基本的 python 培训，得到了一些代码并将它们组合成一个，但我无法完成它。如果有人帮助我，我将不胜感激。错误发生在 vFtp.close() 命令之后。在假设开始同时下载的部分。

from multiprocessing import Pool
import pysftp
import os

vHost='10.11.12.13'
vLogin='admin'
vPwd='pass1234'
vFtpPath='/export/home/'

os.chdir('d:/test/')
os.getcwd()

cnopts=pysftp.CnOpts()
cnopts.hostkeys = None

vFtp=pysftp.Connection(vHost,username=vLogin,password=vPwd,cnopts=cnopts)
vFtp.cwd(vFtpPath)
vObjectList=vFtp.listdir()
vFileList=[]
vFoldList=[]

for vObject in vObjectList:
    vType=str(vFtp.lstat(vObject))[:1]
    if vType!='d': 
        vFileList.append(vObject)
    else:   
        vFoldList.append(vObject)

vFtp.close()

def fDownload(vFileAux):
    vFtpAux=pysftp.Connection(vHost,username=vLogin,password=vPwd,cnopts=cnopts)
    vFtpAux.cwd(vFtpPath)
    vFtpAux.get(vFileAux,preserve_mtime=True)
    vFtpAux.close()

if __name__ == "__main__":
    vPool=Pool(3)
    vPool.map(fDownload,vFileList)  

Answer 1

您似乎正在尝试获取文件列表，然后使用多个进程同时下载它们。

不要手动检查文件，而是尝试在连接对象上使用walktree 方法：pysftp walktree

这是我在 Python 3.5 中制作的一个工作示例。我只是使用本地 ftp 服务器和小文件，所以我模拟了下载延迟。更改 max_workers 参数以设置同时下载的数量。

"""Demo using sftp to download files simultaneously."""
import pysftp
import os
from concurrent.futures import ProcessPoolExecutor
import time


def do_nothing(s):
    """
    Using this as the callback for directories and unknown items found
    using walktree.
    """
    pass


def download(file):
    """
    Simulates a 1-second download.
    """
    with pysftp.Connection(
            host='convox', username='abc', private_key='/home/abc/test') as sftp:

        time.sleep(1)
        print('Downloading {}'.format(file))
        sftp.get(file)


def get_list_of_files(remote_dir):
    """
    Walks remote directory tree and returns list of files.
    """
    with pysftp.Connection(
            host='convox', username='abc', private_key='/home/abc/test') as sftp:

        files = []

        # if this finds a file it will send the filename to the file callback
        # which in this case just appends to the 'files' list
        sftp.walktree(remote_dir, fcallback=files.append,
                      dcallback=do_nothing, ucallback=do_nothing)

    return files

if __name__ == '__main__':
    remote_dir = '/home/abc/remoteftp/'
    download_target = '/home/abc/localftp/'

    # if you don't specify a localpath in sftp.get then it just downloads to
    # the os cwd, so set it here
    os.chdir(download_target)

    files = get_list_of_files(remote_dir)
    pool = ProcessPoolExecutor(max_workers=4)
    pool.map(download, files)

编辑：ProcessPoolExecutor 用于在多个 cpu 内核上运行某些东西，并且会受到您的处理器的限制。对于下载等网络任务，您可以改用线程。在上面的代码中，这只是一个变化：而不是ProcessPoolExecutor 导入并使用ThreadPoolExecutor。那你就可以多用max_workers了。