python 线程池

#coding:utf8
from threading import Thread,Lock
from Queue import Queue,Empty

class Worker(Thread):
    def __init__(self,threadPool):
        Thread.__init__(self)
        self.threadPool = threadPool
        self.daemon = True
        self.state = None
        self.start()    #启动线程

    def stop(self):
        self.state = 'STOP'

    def run(self):
        while True:
            if self.state == 'STOP':
                break
            #获取队列任务
            try:
                func,args,kargs = self.threadPool.getTask(timeout = 1)
            except Empty:
                continue
            #保存结果
            #try:
            result = func(*args,**kargs)
            if result:
                self.threadPool.putTaskResult(result)#存储到结果队列
            self.threadPool.taskDone()
            #except:
            #   print 'error'

class ThreadPool(object):
    def __init__(self,threadNum):
        self.pool = [] #线程池
        self.threadNum = threadNum #线程数
        self.lock = Lock()#线程锁
        self.taskQueue = Queue()#任务队列
        self.resultQueue = Queue()#结果队列

    def startThreads(self):#开启threadNum个线程
        for i in range(self.threadNum):
            self.pool.append(Worker(self))

    def stopThreads(self):
        for thread in self.pool:
            thread.stop()
            thread.join()
        del self.pool[:]

    def putTask(self,func,*args,**kargs):#放入任务队列
        self.taskQueue.put((func,args,kargs))

    def getTask(self,*args,**kargs):
        task = self.taskQueue.get(*args,**kargs)
        return task
    
    def taskJoin(self, *args, **kargs):
        self.taskQueue.join()

    def taskDone(self, *args, **kargs):
        self.taskQueue.task_done()

    def putTaskResult(self, *args):
        self.resultQueue.put(args)

    def getTaskResult(self, *args, **kargs):
        return self.resultQueue.get(*args, **kargs) 

这是参考别人的线程池的代码。https://github.com/lvyaojia/crawler/blob/master/threadPool.py

然后自己的那个爬虫修改如下：

import urllib2
import re
from pyquery import PyQuery as pq
from lxml import etree
import urlparse
import time
from threadPool import ThreadPool

allUrls = set()
allMails = set()
urlsDownlist = []

class mailCrawler:
    def __init__(self,mailExpression,start_url,maxcount,threadNum):   
        ''' mailExpressoin 邮箱的正则表达式;
        start_url开始邮箱；
        maxcount最大数量'''
        self.mailpattern = re.compile(mailExpression)
        self.maxcount = maxcount
        self.htmlcount = 0
        self.UrlsQlist = []#url queue 实现广度优先
        self.url = start_url
        self.threadPool = ThreadPool(threadNum)

    
    def url_normal(self,url):
        '''url 规范化 '''
        scheme,netloc,path,query = urlparse.urlsplit(url)[:4]
        netloc = netloc.lower()

        url.encode("utf-8")

        if path:
            path = re.sub('/{2,}','/',path)#去除url中的重复/
            path = re.sub(r'\.$','',path)#去除url中结尾多余的点
            path = re.sub('/$','',path)#去除url中结尾多余的/
            path = re.sub('\s','',path)#取出url中的空格
        if query:
            return '%s://%s%s?%s' % (scheme,netloc,path or '/',query)
        else:
            return '%s://%s%s' % (scheme,netloc,path)

    def geturls(self,data):
        '''解析html中的url'''
        urls = set()
        if data:  
            d = pq(data)
            label_a = d.find('a')#用pyquery库去找到 a 标签.
            if label_a:
                label_a_href = d('a').map(lambda i,e:pq(e)('a').attr('href'))
                for u in label_a_href:
                    if u[0:10]!="javascript" and u[0:6]!="mailto" :  
                        if u[0:4] == "http":
                            normal_url = self.url_normal(u)
                            urls.add(normal_url)
                        else:
                            normal_url = self.url_normal(self.url + u)
                            urls.add(normal_url)              
            return urls
        else:
            return None
        
    def gethtml(self,url):
        '''下载html  5s超时'''
        try:
            fp = urllib2.urlopen(url,None,5)
        except:
            print "urllib2.urlopen error  or timeout"
            return None
        else:
            try:
                mybytes =fp.read()
            except:
                print "fp.read() error"
                return None
            else:             
                fp.close()
                return mybytes
            
    def savehtmlAndmails(self,pagecontent,htmlcount,url):
        '''保存html文件 '''
        global allMails
        if pagecontent != None:
            f = open("E:/py/crawler/html/"+str(htmlcount)+".html","w")
            f.write(pagecontent)
            f.close()
            '''将抓取到的url存放到 allmails中 ，set去重复'''
            mailResult = self.mailpattern.findall(pagecontent)
            mailResultset = set(mailResult)
            if mailResultset:
                allMails.update(mailResultset)
        else:
            f = open("E:/py/crawler/html/"+str(htmlcount)+"error"+".html","w")
            try:
                f.write(url)
            except:
                f.write("encode error")
            f.close()
            
    def BFS(self):
        '''用队列实现广度优先，爬取url '''
        global allUrls
        global urlsDownlist
        self.threadPool.startThreads()#开始多线程
        allUrls.add(self.url)
        self.UrlsQlist = list(allUrls)
        while self.htmlcount < self.maxcount : #数量小于最大值
            tempUrl = self.UrlsQlist.pop(0)# the queue
            print tempUrl
            urlsDownlist.append(tempUrl)
            myWebStr = self.gethtml(tempUrl)
            #self.savehtml(myWebStr,self.htmlcount,tempUrl)
            #self.savemails(myWebStr)
            self.threadPool.putTask(self.savehtmlAndmails,myWebStr,self.htmlcount,tempUrl)
            firstUrls_set = self.geturls(myWebStr)#初始页面的处理
            if firstUrls_set != None:
                for u in firstUrls_set:
                    if u not in allUrls:
                        allUrls.add(u)
                        self.UrlsQlist.append(u)       
            self.htmlcount = self.htmlcount + 1
        self.threadPool.stopThreads()#结束多线程   

def main():
    reg = r'[A-Za-z0-9_]+@(?:[A-Za-z0-9]+\.)+[A-Za-z]+'
    url = "http://www.163.com"
    count = 100
    threadNum = 10
    fmails = open("E:/py/crawler/mailresult.txt","a")
    furls = open("E:/py/crawler/urlresult.txt","a")
    fdownUrls = open("E:/py/crawler/urlDownresult.txt","a")
    newcrawler = mailCrawler(reg,url,count,threadNum)
    newcrawler.BFS()
    for u in allMails:
        try:
            fmails.write(u)
            fmails.write('\n')
        except:
            continue
    for u in allUrls:
        try:
            furls.write(u)
            furls.write('\n')
        except:
            continue
    for u in urlsDownlist:
        try:
            fdownUrls.write(u)
            fdownUrls.write('\n')
        except:
            continue
    fmails.close()
    furls.close()
    fdownUrls.close()

if __name__ == '__main__':
    main()

我主要是把写文件交给了线程处理。

　　经测试，发现速度根本没有变化。。。冏。。。。

始终觉得根本没有多线程，还没有最初的那个稳定。。。

不知道 从何处下手了。。。