python 多线程示例3

1. 该代码来自：

import threading
import urllib.request
import time
import functools

UPDATE_INTERVAL=0.01

class URLThread(threading.Thread):
    def __init__(self,url):
        super(URLThread,self).__init__()
        self.url=url
        self.response="No Content"

    def run(self):
        headers={'User-Agent':"Mozilla/5.0 (Windows NT 6.1; WOW64; rv:26.0) Gecko/20100101 Firefox/26.0"}
        url=urllib.request.Request(self.url,headers=headers)
        self.request=urllib.request.urlopen(self.url).read()
        self.response=self.request.decode()

def multi_get(uries,timeout=2.0):
    def alive_count(lst):
        alive=map(lambda x: 1 if x.isAlive() else 0 ,lst)
        return functools.reduce(lambda a,b: a+b, alive)

    threads=[URLThread(uri) for uri in uries]
    for thread in threads:
        thread.start()

    while alive_count(threads) > 0 and timeout > 0.0:
        timeout=timeout-UPDATE_INTERVAL
        time.sleep(UPDATE_INTERVAL)
    return [(x.url,x.response) for x in threads]

def run_multi_get():
    sites=["","","",""]
    requests=multi_get(sites,timeout=2.5)
    with open("site","w") as ofile:
        for request in requests:
            output=":".join(request)
            ofile.write(output+"\n")


run_multi_get()

这段代码有意思的函数, 挺有意思的用法，return 当前alive的 thread 的数目。

    def alive_count(lst):
        alive=map(lambda x: 1 if x.isAlive() else 0 ,lst)
        return functools.reduce(lambda a,b: a+b, alive)

如果让我写，我就直接这么写了。

def multi_get(uries):
    threads=[URLThread(uri) for uri in uries]

    for thread in threads:
        thread.start()

    for thread in threads:
        thread.join()

    return [(x.url,x.response) for x in threads]

2.  代码出自: 
该脚本，输入多个目录（小于4个），它对各个目录，各开一个线程统计目录下的文件数目。该脚本hardcode开4个线程

import os,time
import threading
import queue
class WorkerThread(threading.Thread):
    """ A worker thread that takes directory names from a queue, finds all
    files in them recursively and reports the result.
    Input is done by placing directory names (as strings) into the
    Queue passed in dir_q.
    Output is done by placing tuples into the Queue passed in result_q.
    Each tuple is (thread name, dirname, [list of files]).
    Ask the thread to stop by calling its join() method.
    """
    def __init__(self,dir_q,result_q):
        super(WorkerThread,self).__init__()
        self.dir_q=dir_q
        self.result_q=result_q
        self.stoprequest=threading.Event()

    def run(self):
        # As long as we weren't asked to stop, try to take new tasks from the
        # queue. The tasks are taken with a blocking 'get', so no CPU
        # cycles are wasted while waiting.
        # Also, 'get' is given a timeout, so stoprequest is always checked,
        # even if there's nothing in the queue.
        while not self.stoprequest.isSet():
            try:
                dirname=self.dir_q.get(True,0.05)
                filename=list(self.__files_in_dir(dirname))
                self.result_q.put((self.name,dirname,filename))
            except queue.Empty:
                continue

    def join(self,timeout=None):
        self.stoprequest.set()
        super(WorkerThread,self).join(timeout)

    def __files_in_dir(self,dirname):
        #Given a directory name, yields the names of all files (not dirs)
        #contained in this directory and its sub-directories.
        for path,dirs,files in os.walk(dirname):
            for file in files:
                yield os.path.join(path,file)

def utilize_WorkerThread(args):
    #create a single input and single output queue for all threads
    dir_q=queue.Queue()
    result_q=queue.Queue()

    #create the thread pool
    pool=[WorkerThread(dir_q=dir_q,result_q=result_q) for i in range(4)]

    #start all threads
    for thread in pool:
        thread.start()

    #give the workers some work todo
    work_count=0
    for dir in args:
        if os.path.exists(dir):
            work_count+=1
            dir_q.put(dir)

    print("Assigned {} dirs to worker".format(work_count))

    #now get all results
    while work_count > 0:
        #blocking get from a Queue
        result=result_q.get()
        print("From thread {0}: {1} files founded in dir {2}".format(result[0],len(result[2]),result[1]))
        work_count-=1

    #ask threads to die and wait for them to do it
    for thread in pool:
        thread.join()


import sys
utilize_WorkerThread(sys.argv[1:])

3. 是个thread ping 的例子。
对每一个IP 开一个线程去ping 

def ping_thread():
    class ip_check(threading.Thread):

        def __init__(self,ip):
            super(ip_check,self).__init__()
            self.ip=ip
            self.__successful_ping=-1
        def run(self):
            ping_out=os.popen("ping -q -c2 {}".format(self.ip))
            while True:
                line=ping_out.readline()
                if not line:
                    break
                n_received=re.findall(received_packages,line)
                if n_received:
                    self.__successful_ping=int(n_received[0])
        def status(self):
            if self.__successful_ping==0:
                return "no reponse"
            elif self.__successful_ping==1:
                return "alive,but 50% package loss"
            elif self.__successful_ping==2:
                return "alive"
            else:
                return "should not occur"

    received_packages = re.compile(r"(\d) received")

    check_results=[]
    for suffix in range(100,105):
        ip="192.168.0."+str(suffix)
        current=ip_check(ip)
        check_results.append(current)
        current.start()

    for el in check_results:
        el.join()
        logging.debug("status from {0} is {1}".format(el.ip,el.status()))