多线程分段下载研究的python实现（一）

我一直对下载文件比较感兴趣。现在我下载文件大部分是用迅雷，但迅雷也有一些不如意的地方，内存占用大，一些不必要的功能太多，不可定制。尤其是最后一点。现在有些下载对useragent，cookie,auth等要求很多，迅雷这时就无能为力了。最好能有一个在后台运行的download　server，以较小的内存代价运行，界面通过访问本地的端口显示。下面对其中可能涉及的一些技术进行探讨。在python下实现下载比较容易。以下载http://nodejs.org/dist/v0.10.12/node.exe为例：

import urllib2
urllib2.urlopen('http://nodejs.org/dist/v0.10.12/node.exe')

但在网速不给力的情况下，这样下载就会花很长时间，分段下载是个可行的方法。

首先要得到文件的大小：

import httpliub
def _getinfo(host,path):
        conn=httplib.HTTPConnection(host)
        conn.request('HEAD',path)
        res= conn.getresponse()
        if res.status==httplib.OK:
            result=res.getheader('content-length')
        conn.close()
        return int(result)

然后分段下载：

CHUNK=16*1024
def _getpart(host,path,fname,start,end):
        current=start
        headers={'Range':'bytes=%s-%s' % (start, end)}
        f=open(fname,'wb')
        conn=httplib.HTTPConnection(host)
        conn.request('GET',path,'',headers)
        print 'No. %d begin to download range bytes=%s-%s ' % (start, end)
        res=conn.getresponse()
        while True :
            chunk=res.read(CHUNK)
            if not chunk:
                conn.close()
                f.close()
                break
            f.seek(current)
            f.write(chunk)
            print 'No. %d write at %d'%(pid,current)
            current=current+CHUNK

采取res.read(CHUNK)的方式可以减少内存的占用。

分段下载没有问题，但问题是依次分段下载根本没有解决问题，只有同时分段下载才能提高网速。“同时”意味着多线程，复杂的锁，资源分配实非我辈菜鸟所能掌握的。好在有大神给出了解决方案，用gevent。gevent引入了协程的概念，程序实际上要单线程运行，但在程序内部不停地切换，实现多线程的效果。以较低的代价实现多线程，又叫协程。这与nodejs较相似。但nodejs的学习难度太大，只好用gevent了。

说到gevent，我不得不吐糟一下，资料太混乱了。主页面上的是0.13.8，实际上在github上是1.0rc2。文档中介绍的是0.13.8的，1.0rc2变化巨大，与文档根本对不上。只能一点一点摸索。

引入gevent后的整个下载是这样的。

        temp=urlparse.urlparse(url)
        size=_getinfo(temp.netloc,temp.path)
        num=20
        t=size//num
        p=gevent.pool(15)
        fname=url.split('/')[-1]
        for i in range(num):
            start=t*i
            end=start+t-1
            if i==num-1 :end=size
                    job=gevent.spawn(_getpart,temp.netloc,temp.path,fname,start,end)
        p.add(job)
        p.join()
        print 'downloaded'

同时也要对_getpart进行修改

def _getpart(self,pid,host,path,fname,start,end,**kwargs):
        monkey.patch_socket()
        current=start
        headers={'Range':'bytes=%s-%s' % (start, end)}
        f=self._open(fname)
        conn=self._getconn(host)
        conn.request('GET',path,'',headers)
        print 'No. %d begin to download range bytes=%s-%s ' % (pid ,start, end)
        gevent.sleep(0)
        res=conn.getresponse()
        while True :
            gevent.sleep(0)
            chunk=res.read(self.CHUNK)

            if not chunk:
                self._returnconn(host,conn)
                self._close(fname)
                print 'No. %d download range bytes=%s-%s successed' % (pid ,start, end)
                break
            f.seek(current)
            f.write(chunk)
            print 'No. %d write at %d'%(pid,current)
            current=current+self.CHUNK

在凡是需要等待的地方加入gevent.sleep(0),这相当于通知系统这里要等待，可以干别的了。但这这里有个问题，在没有patch（加入monkey.patch_socket()）时，

所有的协程总是以固定的顺序写入，并不像是并行运行。只有patch后才像是并行。找了许多文档也没有解释清楚。

这样就行了吗，还不行。同一文件被反复打开，关闭。如何解决，引入计数。就是这样：

　#文件句柄记录,｛文件名：[文件句柄，引用计数]｝
　self._filehandle={}

def _open(self,fn):
        if fn in self._filehandle:
            self._filehandle[fn][1]+=1
            return self._filehandle[fn][0]
        else:
            fp=open(fn,'wb')
            self._filehandle[fn]=[fp,1]
            return fp

   def _close(self,fn):
        self._filehandle[fn][1]-=1
        if self._filehandle[fn][1] ==0:
            self._filehandle[fn][0].close()
            del self._filehandle[fn]

以后打开文件时调用_open（），关闭时调用_close()。这样避免了重复打开，重复关闭。

再改进下吧，connection也可以重用。下载时大量的联接都是针对同一网站，重用可以加快下载。策略是req完成后，联接并不马上关闭，而是保存的“联接池”中，当需用联接时，先在联接池中查找，有就重用，没有就先新建一条联接。没用的联接在经过一段时间再关闭。所以联接池并不对联接数量进行控制。

既然是延迟关闭，那么就用到了timer，python自带的timer就可以啊。最终还是没用，因为timer是threading下的，属新开线程，消耗太大。另外仅管timer在文档中说是class但实际上根本不是，坑爹啊，是function，_Timer才是class。

gevent的timer也不太好用，文档没有，不行自造一个吧。要求：在固定时隔内运行一系列函数，函数可动态增减，定时器可开关。，结果就是这个，这是网上抄来的，向大侠表示感谢。

class Timer(object):

    def __init__(self,interval):
        self._interval=interval
        self._callbacks = {}
        self._app=None
        self._flag=False

    def start(self):
        self._flag=True
        if self._app==None:
            self._app = gevent.spawn( self._loop )

    def stop(self):
        self._flag=False
        self._app==None

    def _loop( self ):
        while not self._flag:
            for callback in self._callbacks.value():
                callback()
            gevent.sleep( self.interval )

    def status(self):
        return self._flag

    def add(self, funcname,func, *args, **kwargs ):
        if not funcname in self._callbacks:
            self._callbacks[ funcname ]= callback(func, *args, **kwargs)

    def remove(funcname):
        if funcname in self._callbacks:
            del self._callbackes[funcname]

严格来说这个定时器还是有些隐患，因为在_callbacks中的所有函数执行完成后，才开始计时，如果执行时间较长，或有异常就会出问题。上面提到的callback是对函数的封装，也是向同一位大侠抄来的：

class callback:
    """
    可执行对象
    """
    def __init__( self, func, *args, **kwargs ):
        self.func = func
        self.args = list(args)
        self.kwargs = kwargs

    def __call__( self, *args, **kwargs ):
        for k in self.kwargs:
            if not kwargs.has_key(k):
                kwargs[k] = self.kwargs[k]
        args = self.args + list(args)
        return self.func( *args, **kwargs )

    def __str__( self ):
        return "<CallableObject %s %s %s />"%(str(self.func),str(self.args),str(self.kwargs))

这样复用联接就好办了

def _getconn(self,host):
        result=None
        for item in self._conn:
            if item[0]==host:
                result=item[1]
                self._conn.remove(item)
                break
        if result==None:
            result=httplib.HTTPConnection(host)
            result.connect()
        return result

    def _returnconn(self,host,conn):
        self._conn.append([host,conn,time.time()])
        self._timer.add('closeconn',self._closeconn)

    def _closeconn(self):
        limit=time.time()-self.CONNDELAY
        temp=[]
        for item in self._conn:
            if item[2]<limit:
                item.close()
            else:
                temp.append(item)
        self._conn=temp
        if len(self._conn)==0:
            self._timer.remove('closeconn')

三个函数分别是取联接，归还联接，及定时器中的关闭联接。这样还不行，有的重用联接还有抛出异常，responsenotready。问题出在第一个head request上，request一定要read只后联接才能重用。

def _getinfo(self,host,path, **kwargs):
        conn=self._getconn(host)
        conn.request('HEAD',path)
        res= conn.getresponse()
        res.read()
        if res.status==httplib.OK:
            result=res.getheader('content-length')
        self._returnconn(host,conn)
        return int(result)

最后再把整个封装成一个类，可以随时增加下载。下面是完整的代码：

 from gevent import monkey
 from gevent.pool import Pool
 from gevent import greenlet
 import urlparse,httplib,gevent,time
 from gevent.core import timer

 #封装callback
 class callback:

     def __init__( self, func, *args, **kwargs ):
         self.func = func
         self.args = list(args)
         self.kwargs = kwargs

     def __call__( self, *args, **kwargs ):
         for k in self.kwargs:
             if not kwargs.has_key(k):
                 kwargs[k] = self.kwargs[k]
         args = self.args + list(args)
         return self.func( *args, **kwargs )

     def __str__( self ):
         return "<CallableObject %s %s %s />"%(str(self.func),str(self.args),str(self.kwargs))

 #定时器类
 class Timer(object):

     def __init__(self,interval):
         self._interval=interval
         self._callbacks = {}
         self._app=None
         self._flag=False

     def start(self):
         self._flag=True
         if self._app==None:
             self._app = gevent.spawn( self._loop )

     def stop(self):
         self._flag=False
         self._app==None

     def _loop( self ):
         while not self._flag:
             for callback in self._callbacks.value():
                 callback()
             gevent.sleep( self.interval )

     def status(self):
         return self._flag

     def add(self, funcname,func, *args, **kwargs ):
         if not funcname in self._callbacks:
             self._callbacks[ funcname ]= callback(func, *args, **kwargs)

     def remove(funcname):
         if funcname in self._callbacks:
             del self._callbackes[funcname]

 #服务器类，
 class server(object):

     def __init__(self):

         #读写块大小
         self.CHUNK=16*1024
         #下载协程池大小
         self.POOLSIZE=15
         #下载超时（S）
         self.TIMEOUT=10
         #下载重试次数
         self.RETRY=10
         #连接空闲后，关闭前的时间
         self.CONNDELAY=3
         #计时器扫描间隔
         self.INTERVAL=1.0
         #文件句柄记录,｛文件名：[文件句柄，引用计数]｝
         self._filehandle={}
         #联接记录，[host,conn,time.time()]
         self._conn=[]
         self._pool=Pool(self.POOLSIZE)
         self._timer=Timer(self.INTERVAL)

     #建立联接
     def _getconn(self,host):
         result=None
         for item in self._conn:
             if item[0]==host:
                 result=item[1]
                 self._conn.remove(item)
                 break
         if result==None:
             result=httplib.HTTPConnection(host)
             result.connect()
         return result

     #将富余连接置入self._conn
     def _returnconn(self,host,conn):
         self._conn.append([host,conn,time.time()])
         self._timer.add('closeconn',self._closeconn)

     #关闭超时连接
     def _closeconn(self):
         limit=time.time()-self.CONNDELAY
         temp=[]
         for item in self._conn:
             if item[2]<limit:
                 item.close()
             else:
                 temp.append(item)
         self._conn=temp
         if len(self._conn)==0:
             self._timer.remove('closeconn')

     #记录文件句柄
     def _open(self,fn):
         if fn in self._filehandle:
             self._filehandle[fn][1]+=1
             return self._filehandle[fn][0]
         else:
             fp=open(fn,'wb')
             self._filehandle[fn]=[fp,1]
             return fp

     #按引用计数关闭文件句柄　
     def _close(self,fn):
         self._filehandle[fn][1]-=1
         if self._filehandle[fn][1] ==0:
             self._filehandle[fn][0].close()
             del self._filehandle[fn]

     #取得下载文件信息
     def _getinfo(self,host,path, **kwargs):
         conn=self._getconn(host)
         conn.request('HEAD',path)
         res= conn.getresponse()
         res.read()
         if res.status==httplib.OK:
             result=res.getheader('content-length')
         self._returnconn(host,conn)
         return int(result)

     #分段下载
     def _getpart(self,pid,host,path,fname,start,end,**kwargs):
         monkey.patch_socket()
         current=start
         headers={'Range':'bytes=%s-%s' % (start, end)}
         f=self._open(fname)
         conn=self._getconn(host)
         conn.request('GET',path,'',headers)
         print 'No. %d begin to download range bytes=%s-%s ' % (pid ,start, end)
         gevent.sleep(0)
         res=conn.getresponse()
         while True :
             gevent.sleep(0)
             chunk=res.read(self.CHUNK)

             if not chunk:
                 self._returnconn(host,conn)
                 self._close(fname)
                 print 'No. %d download range bytes=%s-%s successed' % (pid ,start, end)
                 break
             f.seek(current)
             f.write(chunk)
             print 'No. %d write at %d'%(pid,current)
             current=current+self.CHUNK

     #添加下载任务
     def add(self,url):
         temp=urlparse.urlparse(url)
         size=self._getinfo(temp.netloc,temp.path)
         num=20
         t=size//num
         fname=url.split('/')[-1]
         for i in range(num):
             start=t*i
             end=start+t-1
             if i==num-1 :end=size
             job=gevent.spawn(self._getpart,i,temp.netloc,temp.path,fname,start,end)
             self._pool.add(job)
         self._pool.join()
         print 'downloaded'

这个东西还存在很多问题，没有实现持久化，不能控制具体下载线程，没有处理重定向、超时等，下一步进一步研究。欢迎各位达人指正，拍砖。