Python模块整理(四)：线程模块threading

 

参考网络以及python书籍整理
一、Python threading三种调用方式介绍：
Thread 是threading模块中最重要的类之一，可以使用它来创建线程。
第一种方式:创建一个threading.Thread()的实例对象，给它一个函数。在它的初始化函数（__init__）中将可调用对象作为参数传入
第二种方式:创建一个threading.Thread的实例，传给它一个可调用类对象，类中使用__call__()函数调用函数
第三种方式:是通过继承Thread类，重写它的run方法；
第一种和第三种常用。
实例可参考：http://tuoxie174.blog.51cto.com/1446064/442162
二、实际简单使用
这里使用第三种方式:是通过继承Thread类，重写它的run方法
#下面的例子treadTest.py创建一个threading.Thread的一个子类KissThread，这子类KissThread重写了超类threading.Thread的run方法
#使用时创建这个子类的实例对象。然后调用该实例的start()启动run()函数
#run方法和start方法:它们都是从Thread继承而来的，run()方法将在线程开启后执行，可以把相关的逻辑写到run方法中（通常把run方法称
#为活动[Activity]）；start()方法用于启动线程。
#!/usr/bin/env python
import threading
import time
count=1

class KissThread(threading.Thread):
        def run(self):
                global count
                print "Thread # %s:Pretending to do stuff" % count
                count+=1
                time.sleep(2)
                print "done with stuff"


for t in range(5):
        KissThread().start()
                   
#ping单线程实例    common.py
#!/usr/bin/env python
import subprocess
import time

IP_LIST=['qq.com','163.com','sohu.com']
cmd_stub='ping -c 5 %s'

def do_ping(addr):
        print time.asctime(),"DOING PING FOR",addr
        cmd=cmd_stub % addr
        return subprocess.Popen(cmd,shell=True,stdout=subprocess.PIPE)

z=[]
for ip in IP_LIST:
        p=do_ping(ip)    #对象
        z.append((p,ip)) #将p这个对象和其产生对象的参数ip作为一个元组添加到列表中
       
for p,ip in z:
        print time.asctime(),"WAITING PING FOR",ip
        p.wait()
        print time.asctime(),ip,"RETURN",p.returncode
三、线程化的ping扫描
这里使用第一种方式:创建一个threading.Thread()的实例对象，给它一个函数
创建一个threading.Thread()的实例，给它一个函数
参数group是预留的，用于将来扩展；
参数target是一个可调用对象（也称为活动[activity]），在线程启动后执行；
参数name是线程的名字。默认值为“Thread-N“，N是一个数字。
参数args和kwargs分别表示调用target时的参数列表和关键字参数。
#vim threadingping.py
#!/usr/bin/env python
import subprocess
from threading import Thread
from Queue import Queue

num_thread=3  #定义线程的数量
queue=Queue() #创建队列实例
ips=['192.168.1.100','192.168.1.110','192.168.1.120','192.168.1.130','192.168.1.200']

def pinger(i,q):
        while True:
                ip=q.get() #获取Queue队列传过来的ip，队列使用队列实例queue.put(ip)传入ip，通过q.get() 获得
                print "Thread %s:Pinging %s" %(i,ip)
                ret=subprocess.call("ping -c 1 %s" % ip,shell=True,stdout=open('/dev/null','w'),stderr=subprocess.STDOUT)
                #调用子进程执行命令，获取退出状态。不能使用subprocess.Popen也可以
                if ret==0:
                        print "%s:is alive" % ip
                else:
                        print "%s:did not respond" % ip
                q.task_done() #告诉queue.join()已完成队列中提取元组的工作

for i in range(num_thread):#各线程开始工作
        worker=Thread(target=pinger,args=(i,queue)) #创建一个threading.Thread()的实例，给它一个函数以及函数的参数
        worker.setDaemon(True)    #在start方法被调用之前如果没有进行设置，程序会不定期挂起。
        worker.start()     #开始线程的工作，没有设置程序会挂起，不会开始线程的工作，因为pinger程序是while True循环

for ip in ips:
        queue.put(ip)    #将IP放入队列中。函数中使用q.get(ip)获取

print "Main Thread Waiting"
queue.join()    #防止主线程在其他线程获得机会完成队列中任务之前从程序中退出。
print "Done"
 
四、多队列和多线程池
# vim multhreading.py
#!/usr/bin/env python
import subprocess
import re
from threading import Thread
from Queue import Queue

num_ping_threads=3
num_arp_threads=3
in_queue=Queue()
out_queue=Queue()
ips=['192.168.1.100','192.168.1.110','192.168.1.120','192.168.1.130','192.168.1.200']

def pinger(i,iq,oq):
        while True:
                ip=iq.get()
                print "Thread %s:Pinging %s" %(i,ip)
                ret=subprocess.call("ping -c 1 %s" % ip,shell=True,stdout=open('/dev/null','w'),stderr=subprocess.STDOUT)
                if ret==0:
                        #print "%s:is alive" % ip
                        oq.put(ip) #将能ping通的ip吐给下另外一个队列
                else:
                        print "%s:did not respond" % ip
                iq.task_done()# 告诉iq.join()已完成队列中提取元组的工

def arping(i,oq):
        while True:
                ip=oq.get() #从pinger函数中吐出的oq队列里获取IP
                p=subprocess.Popen("arping -c 1 %s" % ip,shell=True,stdout=subprocess.PIPE)
                #需要获取子进程返回的内容，而不是简单的退出状态
                out=p.read()#返回的内容读出来
                result=out.split()
                pattern=re.compile(":") #正则编译匹配
                macaddr=None
                for item in result:
                        if re.search(pattern,item):#查找到匹配的就赋值给macaddr
                                macaddr=item
                print "IP Address:%s| Mac Address:%s" % (ip,macaddr)
                oq.task_done()# 告诉oq.join()已完成队列中提取元组的工

for ip in ips:
        in_queue.put(ip)

for i in range(num_ping_threads):
        worker=Thread(target=pinger,args=(i,in_queue,out_queue)) #创建一个threading.Thread()的实例，给它一个函数以及函数的参数
        worker.setDaemon(True)    #在start方法被调用之前如果没有进行设置，程序会不定期挂起
        worker.start()     #开始线程的工作，没有设置程序会挂起，不会开始线程的工作

for i in range(num_arp_threads):
        worker=Thread(target=arping,args=(i,out_queue))
        worker.setDaemon(True)
        worker.start()

print "Main Thread Waiting"
in_queue.join() #防止主线程在其他线程获得机会完成队列中任务之前从程序中退出。
out_queue.join()
print "Done"

五、使用threading.Timer线程延迟
Timer(延迟时间,函数)
threading.Timer是threading.Thread的子类，可以在指定时间间隔后执行某个操作。下面是Python手册上提供的一个例子：
    def hello(): 
        print "hello, world" 
    t = Timer(3, hello) 
    t.start() # 3秒钟之后执行hello函数
具体实例
# vim delay_threading.py
#!/usr/bin/env python
import sys
import time
import copy
from threading import Timer

if len(sys.argv)!=2:
        print "Must enter an interval"
        sys.exit(1)
       
def hello():
        print "Hello,I just got called after a %s sec delay" % call_time
       
delay=sys.argv[1]
call_time=copy.copy(delay) #copy是因为delay要倒计时递减
t=Timer(int(delay),hello)
t.start()

print "waiting %s seconds to run function" % delay
for x in range(int(delay)):
        print "Main program is still running for %s more sec" % delay
        delay=int(delay)-1
        time.sleep(1)
语句有点混乱重新改写过
# cp delay_threading.py delay_threading_new.py
#vim delay_threading_new.py
#!/usr/bin/env python
"""
USAGE:
        delay_threading_new.py intarg
"""

import sys
import time
import copy
from threading import Timer


def hello():
        print "Hello,I just got called after a %s sec delay" % call_time

if __name__=='__main__':
        if len(sys.argv)!=2:
                print __doc__
        else:
                try:
                        delay=sys.argv[1]
                        t=Timer(int(delay),hello)
                        call_time=copy.copy(delay)
                except  ValueError:
                        print __doc__
                        sys.exit(1)
                else:
                        t.start()
                        print "waiting %s seconds to run function" % delay
                        for x in range(int(delay)):
                                print "Main program is still running for %s more sec" % delay
                                delay=int(delay)-1
                                time.sleep(1)
           
六、Thread类其他常用方法与属性：
1、一般方法和属性 【类Thread的方法】
Thread.getName()
Thread.setName()
Thread.name
用于获取和设置线程的名称。
Thread.ident
获取线程的标识符。线程标识符是一个非零整数，只有在调用了start()方法之后该属性才有效，否则它只返回None。
Thread.is_alive()
Thread.isAlive()
判断线程是否是激活的（alive）。从调用start()方法启动线程，到run()方法执行完毕或遇到未处理异常而中断 这段时间内，线程是激活的。
Thread.join([timeout])
2、Thread.join 【类Thread的方法】
调用Thread.join将会使主调线程堵塞，直到被调用线程运行结束或超时。参数timeout是一个数值类型，表示超时时间，如果未提供该参数，那么主调线程将一直堵塞到被调线程结束。下面举个例子说明join()的使用：
import threading, time 
def doWaiting(): 
    print 'start waiting:', time.strftime('%H:%M:%S') 
    time.sleep(3) 
    print 'stop waiting', time.strftime('%H:%M:%S') 
thread1 = threading.Thread(target = doWaiting) 
thread1.start() 
time.sleep(1)  #确保线程thread1已经启动 
print 'start join' 
thread1.join()  #将一直堵塞，直到thread1运行结束。 
print 'end join'
            
3、锁相关方法 【模块threading的方法】
#使用队列Queue就可以避免使用锁的麻烦
threading.RLock和threading.Lock
在threading模块中，定义两种类型的琐：threading.Lock和threading.RLock。它们之间有一点细微的区别，通过比较下面两段代码来说明：
import threading 
lock = threading.Lock() #Lock对象 
lock.acquire() 
lock.acquire()  #产生了死琐。 
lock.release() 
lock.release() 

import threading 
rLock = threading.RLock()  #RLock对象 
rLock.acquire() 
rLock.acquire() #在同一线程内，程序不会堵塞。 
rLock.release() 
rLock.release() 

这两种琐的主要区别是：RLock允许在同一线程中被多次acquire。而Lock却不允许这种情况。注意：如果使用RLock，那么acquire和release必须成对出现，即调用了n次acquire，必须调用n次的release才能真正释放所占用的琐。

4、threading.Condition 【模块threading的方法】
threading.Condition
可以把Condiftion理解为一把高级的琐，它提供了比Lock, RLock更高级的功能，允许我们能够控制复杂的线程同步问题。threadiong.Condition在内部维护一个琐对象（默认是RLock），可以在创建Condigtion对象的时候把琐对象作为参数传入。Condition也提供了acquire, release方法，其含义与琐的acquire, release方法一致，其实它只是简单的调用内部琐对象的对应的方法而已。Condition还提供了如下方法(特别要注意：这些方法只有在占用琐(acquire)之后才能调用，否则将会报RuntimeError异常。)：
Condition.wait([timeout]): 
wait方法释放内部所占用的琐，同时线程被挂起，直至接收到通知被唤醒或超时（如果提供了timeout参数的话）。当线程被唤醒并重新占有琐的时候，程序才会继续执行下去。
Condition.notify():
唤醒一个挂起的线程（如果存在挂起的线程）。注意：notify()方法不会释放所占用的琐。
Condition.notify_all()
Condition.notifyAll()
唤醒所有挂起的线程（如果存在挂起的线程）。注意：这些方法不会释放所占用的琐。

5、threading.Event 【模块threading的方法】
Event实现与Condition类似的功能，不过比Condition简单一点。它通过维护内部的标识符来实现线程间的同步问题。（threading.Event和.NET中的System.Threading.ManualResetEvent类实现同样的功能。）
Event.wait([timeout])
堵塞线程，直到Event对象内部标识位被设为True或超时（如果提供了参数timeout）。
Event.set()
将标识位设为Ture
Event.clear()
将标识伴设为False。
Event.isSet()
判断标识位是否为Ture。

6、threading模块其他方法：
threading.active_count()
threading.activeCount()
获取当前活动的(alive)线程的个数。
threading.current_thread()
threading.currentThread()
 　　获取当前的线程对象（Thread object）。
threading.enumerate()
 　　获取当前所有活动线程的列表。
threading.settrace(func)
设置一个跟踪函数，用于在run()执行之前被调用。
threading.setprofile(func)
设置一个跟踪函数，用于在run()执行完毕之后调用。

参考：http://blog.csdn.net/JGood/article/details/4305604
参考：unix/linux python系统管理

 

本文出处：

http://ipseek.blog.51cto.com/1041109/808818