多进程下载网络拥堵问题咨询

我需要下载约 10 万个小文件（每个 1M 左右。）

写了一个脚本，开 8 个进程，每个进程用 10 个线程下载。

问题是: 无论如何修改进程，线程数，下载速度都没有太大变化，每秒 4 个左右。而且本机的网络会很卡，打开网页很慢，,ping 192.168.1.1 （网关）都要几百毫秒。这是为什么？应该怎么优化？同路由器链接的其他机器，网络不受影响。我的机器配置： i7-3667u，256GB SSD， 8G 内存，ubuntu 14.04 64 位系统

64 bytes from 192.168.1.1: icmp_seq=372 ttl=64 time=210 ms
64 bytes from 192.168.1.1: icmp_seq=373 ttl=64 time=136 ms
64 bytes from 192.168.1.1: icmp_seq=374 ttl=64 time=166 ms
64 bytes from 192.168.1.1: icmp_seq=375 ttl=64 time=139 ms
64 bytes from 192.168.1.1: icmp_seq=376 ttl=64 time=369 ms
64 bytes from 192.168.1.1: icmp_seq=377 ttl=64 time=537 ms
64 bytes from 192.168.1.1: icmp_seq=378 ttl=64 time=536 ms
64 bytes from 192.168.1.1: icmp_seq=379 ttl=64 time=582 ms
64 bytes from 192.168.1.1: icmp_seq=380 ttl=64 time=461 ms
64 bytes from 192.168.1.1: icmp_seq=381 ttl=64 time=363 ms
64 bytes from 192.168.1.1: icmp_seq=382 ttl=64 time=309 ms
64 bytes from 192.168.1.1: icmp_seq=383 ttl=64 time=277 ms
64 bytes from 192.168.1.1: icmp_seq=384 ttl=64 time=312 ms
64 bytes from 192.168.1.1: icmp_seq=385 ttl=64 time=268 ms
64 bytes from 192.168.1.1: icmp_seq=386 ttl=64 time=392 ms
64 bytes from 192.168.1.1: icmp_seq=387 ttl=64 time=400 ms
64 bytes from 192.168.1.1: icmp_seq=388 ttl=64 time=266 ms
64 bytes from 192.168.1.1: icmp_seq=389 ttl=64 time=551 ms
64 bytes from 192.168.1.1: icmp_seq=390 ttl=64 time=382 ms
64 bytes from 192.168.1.1: icmp_seq=391 ttl=64 time=289 ms
64 bytes from 192.168.1.1: icmp_seq=392 ttl=64 time=309 ms
64 bytes from 192.168.1.1: icmp_seq=393 ttl=64 time=445 ms
64 bytes from 192.168.1.1: icmp_seq=394 ttl=64 time=411 ms
64 bytes from 192.168.1.1: icmp_seq=395 ttl=64 time=274 ms
64 bytes from 192.168.1.1: icmp_seq=396 ttl=64 time=262 ms
64 bytes from 192.168.1.1: icmp_seq=397 ttl=64 time=282 ms
64 bytes from 192.168.1.1: icmp_seq=398 ttl=64 time=588 ms
64 bytes from 192.168.1.1: icmp_seq=399 ttl=64 time=614 ms
64 bytes from 192.168.1.1: icmp_seq=400 ttl=64 time=222 ms
64 bytes from 192.168.1.1: icmp_seq=401 ttl=64 time=238 ms
64 bytes from 192.168.1.1: icmp_seq=402 ttl=64 time=335 ms
64 bytes from 192.168.1.1: icmp_seq=403 ttl=64 time=461 ms
64 bytes from 192.168.1.1: icmp_seq=404 ttl=64 time=491 ms

#coding=utf8

import threading

import multiprocessing
import time
import Queue
import os
import re
import requests

#SHARE_Q = Queue.Queue()  #构造一个不限制大小的的队列
SHARE_Q = multiprocessing.Queue()  #构造一个不限制大小的的队列
_WORKER_THREAD_NUM = 2   #设置线程个数
_WORKER_PROCESS_NUM = 8

class MyThread(threading.Thread) :

    def __init__(self, func, q) :
        super(MyThread, self).__init__()
        self.func = func
        self.queue = q

    def run(self) :
        self.func(self.queue)

def worker(q):
    session = requests.Session()
    while not q.empty():
        url, name = q.get() #获得任务
        print "Processing : ", url, name
        with open(name, 'w') as f:
            f.write(session.get(url).content)

def main() :
    global SHARE_Q


for root,dirs,files in os.walk('data'):
    for filespath in files:
        url = os.path.join(root,filespath)
        url = os.path.abspath(url)
        temp = []
        if url.endswith('m3u8'):
            path = os.path.dirname(url)
            new_m3u8_path = os.path.join(path, 'new_video.m3u8')
            content = open(url).read()
            for x in content.split('\n'):
                if x.startswith('http://'):
                    x = x.split('?')[0]
                    _, name = os.path.split(x)
                    save_path = os.path.join(path, name)
                    temp.append(name)
                    if not os.path.exists(save_path) or os.path.getsize(save_path) == 0:
                        #print 'download', x, save_path
                        #os.system('wget %s -O %s' % (x, save_path))
                        SHARE_Q.put([x, save_path])
                        #with open(save_path, 'w') as f:
                            #f.write(requests.get(x).content)
                else:
                    temp.append(x)
            with open(new_m3u8_path, 'w') as f:
                f.write("\n".join(temp))
            #exit()

processs = []
print 'Queue length', SHARE_Q.qsize()

def func(q, max_thread_num):
    threads = []
    for i in xrange(max_thread_num) :
        thread = MyThread(worker, q)
        thread.start()
        threads.append(thread)
    for thread in threads :
        thread.join()

for i in xrange(_WORKER_PROCESS_NUM):
    p = multiprocessing.Process(target=func, args=(SHARE_Q, _WORKER_THREAD_NUM))
    p.start()
    processs.append(p)

while True:
    time.sleep(10)
    print 'Queue length', SHARE_Q.qsize()

for p in processs:
    p.join()

多进程下载网络 拥堵问题咨询

多进程下载网络拥堵问题咨询