python多线程测试

仿照之前自己写的一个perl的多线程的脚本,简单试了下python的多线程模块threading.只能说因为之前习惯了perl的很多用法,刚切换到python非常不习惯,一些小的操作上不熟练.


#!/usr/bin/env python
import threading
import time
import random
import sys
import string
def fun_test(x):
    sleep_time=random.random()*1
    print "args is %s sleep time is %.4f s" % (x,sleep_time)
    time.sleep(sleep_time)
def multi_do(lists,concurrent_max,func):
    concurrent=0
    thread_list=set()
    while( len( lists ) > 0 ):
        if len(thread_list) <= concurrent_max:
            ser=lists.pop()
            pid=threading.Thread(target=func,name=ser,args=(ser,))
            thread_list.add(pid)
            pid.start()
        else:
            alive_threads=set(threading.enumerate())
            join_threads=thread_list-alive_threads
            for job in join_threads:
                print "%s is done"% job.getName()
                job.join()
                thread_list.remove(job)
    while(len(thread_list)>0):
        alive_threads=set(threading.enumerate())
        join_threads=thread_list-alive_threads
        for job in join_threads:
            print "%s is done"% job.getName()
            job.join()
            thread_list.remove(job)
    print "all job have been done"
if __name__ == "__main__":
    try:
        fd=open(sys.argv[1],"r")
        print "open %s ok" % sys.argv[1]
        todo_list=fd.readlines()
        todo_list=map(lambda x: x.strip(),todo_list)
        print "to do is:\n","\n".join(todo_list)
    except IOError:
        print "open file failed"
    fd.close()
    multi_do(todo_list,10,fun_test)
此条目发表在python分类目录。将固定链接加入收藏夹。

发表评论