并行下载与多处理和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。