Python 多线程 -thread threading Queue- 简单学习

Python 多线程 -thread threading Queue- 简单学习

在实际工作过程中，会出现需要并发的做一些事情，例如一台机器测到几千台机器的网络连通性，如果你单线程一台一台测的话，会花费很多的事情，不具有实时性，更不能在变化的时候立刻感知当时网络的状况，这时多线程就是一个很好地选择。python已经给我们封装好了多线程库thread和threading。

thread:比较底层的模块
threading：Higher-level threading interface

ps：建议使用threading模块
- 高级别的threading模块更为先进，对线程的支持更为完善
- 低级别的thread模块同步原语很少
- thread模块对线程什么时候结束完全没有控制，当主线程结束时，所有线程都会强制结束

thread

模块函数

start_new_thread(function, args，kwargs=None): 产生新的线程，args是function的参数，没有时写（），kwargs用来调用这个函数
allocate_lock(): 分配锁，LockType类型
exit(): 让线程退出

LockType的操作

acquire（wait=None）：尝试获取锁
locked（）： 获取了锁返回True，没有返回False
release（）：释放锁

Demo1

$ cat t1.py
import thread
from time import sleep

def a():
    print "a start"
    sleep(2)
    print "a end"

def b():
    print "b start"
    sleep(2)
    print "b end"

def main():
    thread.start_new_thread(a,())
    thread.start_new_thread(b,())
    print "all done"

if __name__ == "__main__":
    main()

$ python t1.py
all done
b start
a start

最终会发现，每一次运行出来的结果都有可能不同，但是绝对不会出现“a end”和“b end”。这是为什么呢，这里没有写让主线程停下来等所有子线程结束后再继续运行的代码，所以main线程在执行完print "all done"就关闭了a和b两个线程。怎么办呢，可以在这里加一个sleep等待子进程执行完毕后再退出。

Demo2： thread -- 多线程的演示 by sleep

$ cat t2.py
import thread
from time import sleep

def a():
    print "a start"
    sleep(2)
    print "a end"

def b():
    print "b start"
    sleep(2)
    print "b end"

def main():
    thread.start_new_thread(a,())
    thread.start_new_thread(b,())
    sleep (4)       ----防止主进程过早退出，加sleep等待子进程执行完毕后再推出
    print "all done"

if __name__ == "__main__":
    main()
$ python t1.py 
b start
a start
a end
b end
all done

但是假设我们不知道子进程执行的时间怎么办，这就是锁的用武之地了。因为使用锁要比使用sleep()函数更为合理。如下所示：

Demo3: thread -- 多线程演示 by lock

实现方式为： 主线程初始化两个锁，分别传给两个函数，两个函数在执行完自己的代码后释放锁，主线程一直在轮询这个锁有没有释放，如果释放了就退出。

def a(lock, nsec):
    print "a starting at :", ctime()
    sleep(nsec)
    lock.release()    -- 执行完之后释放锁
    print "a end", ctime()

def b(lock, nsec):
    print "b starting at :", ctime()
    sleep(nsec)
    lock.release()    -- 执行完之后释放锁
    print "b end", ctime()

def main():
    print "Demo Starting at:", ctime()

    locks = []

    # Initialize lock  -- 主线程先获取两个锁，占为己有
    for i in range(2):
        lock = thread.allocate_lock()
        lock.acquire()
        locks.append(lock)

    # 每个进程分配一个锁
    thread.start_new_thread(a, (locks[0],2))
    thread.start_new_thread(b, (locks[1],4))

    for i in range(2):   #一直在轮询，看锁有没有释放
        while locks[i].locked(): pass 

    print "all done at:", ctime()

最后的结果为：

$ python thread_demo.py 
Demo Starting at: Fri Aug 29 22:03:01 2014
a starting at : Fri Aug 29 22:03:01 2014
b starting at : Fri Aug 29 22:03:01 2014
a end Fri Aug 29 22:03:03 2014
b end Fri Aug 29 22:03:05 2014
all done at: Fri Aug 29 22:03:05 2014

不难发现，thread库的同步机制比较难用，一切都需要主进程来处理。并且没有守护进程，主进程一退，整个世界都会变得很清静。而threading库给我们提供了守护进程。下面就来看看threading的简单用法。

threading

threading提供了Thread类，还提供了很多非常好用的同步机制。感觉重点了解Thread类就可以，多线程，也就是通过Thread类的多个实例。 类的主要方法有：

start（）：开始线程的执行。thread库里里面，是没有办法控制线程的开始的
join（timeout=None）: 等待线程结束，有点类似Demo3中的轮询
run（）：定义线程的功能

感觉上面是比较重要的，立马就会用到的。还有一些其他的：

getName（）：获取线程名
setName（name）：设置线程名
isAlive(): 返回bool 表示线程是否在运行中
activeCount（）：返回运行中的线程数
currentThread（）：返回当前线程对象
enumerate（）：返回当前活动线程的列表
isDaemon(): 返回线程的Daemon标志
setDaemon（daemonic）： 设置线程的Daemon标志，一般在start（）函数前调用
settrace（func）：为所有线程设置跟踪函数
setprofile（func）： 为所有线程设置profile函数

Demo4 -- threading演示

def loop(i, nsec):
    print "thread %d starting at : %s" %(i, ctime())
    sleep(nsec)
    print "thread %d end at : %s" %(i, ctime())

def main():
    threads = []
    loops = [2, 4]
    # 实例化进程
    for i in range(len(loops)):
        t = threading.Thread(target = loop, args = (i, loops[i]))
        threads.append(t)

    for i in range(len(loops)):
        threads[i].start()

    for i in range(len(loops)):
        threads[i].join()

    print "all done"

最后的结果为：

thread 0 starting at : Sun Aug 31 13:31:28 2014
thread 1 starting at : Sun Aug 31 13:31:28 2014
thread 0 end at : Sun Aug 31 13:31:30 2014
thread 1 end at : Sun Aug 31 13:31:32 2014
all done

可见threading可以方便的控制线程的开始，以及等待每个线程的结束，并且也不用设置锁，释放锁，这些都被threading库封装了，相比于thread要更高级一些。在实际的运维工程中，可能会需要多个线程执行相同的任务，这时需要一个任务池。每个线程取任务池中取任务，执行，再取任务，再执行，一直到任务池为空，退出线程。这里就会用到下面要介绍的Queue库。

Queue

Queue模块可以用来实现多线程间通讯，让各个线程共享数据，生产者把货物放到Queue中，供消费者（线程）去使用。在python3中，Queue模块被命名为queue。 Queue的对象有：

Queue.Queue(maxsize=0)： 创建大小为maxsize的FIFO(First In First Out)-Queue对象，如果maxsize不设置，这个队列将是无限的。
Queue.LifoQueue(maxsize=0): 创建先入后出的对象，即栈, 在python2.6中加入 Queue.PriorityQueue(maxsize=0):有优先级的队列，在python2.6中加入

Queue对象的方法有：

qsize（）：返回队列的大小
empty（）：返回队列时候为空
full（）：返回队列是否满
put（item，block=0）：向Queue对象中放数据，block不为0时，会一直等到队列中有控件为止
get（block=0）：，同上，block不为0时，会一直等到队列中有数据为止

Demo5 -- Queue的使用演示：
场景: Queue里面放着一些整数，需要将整数取出，并且睡眠整数大小的时间，下面的demo中，是放了10个1，如果单线程的话需要10s

def work(q):
    while True:
        if q.empty():
            return
        else:
            t = q.get()
            time.sleep(t)

def main():
    q = Queue.Queue()   # 初始化一个Queue对象
    for i in range(10): # 向Queue生产任务
        q.put(1)

    work(q)

if __name__ == "__main__":
    main() 

最后的结果为：

time python threading_demo2.py 

real    0m10.085s
user    0m0.060s
sys 0m0.004s

单线程的话需要花费10s。

下面来通过多线程来处理Queue里面的任务：

def work(q):
    while True:
        if q.empty():
            return
        else:
            t = q.get()
            time.sleep(t)

def main():
    q = Queue.Queue()
    for i in range(10):
        q.put(1)

    thread_num = 10
    threads = []
    for i in range(thread_num):
        t = threading.Thread(target = work, args = (q,)) # args需要输出的是一个元组，如果只有一个参数，后面加，表示元组，否则会报错
        threads.append(t)

    for i in range(thread_num):
        threads[i].start()

    for i in range(thread_num):
        threads[i].join()

看看这下的结果为：

real    0m1.046s
user    0m0.024s
sys     0m0.020s

因此对python多线程，主要学会使用threading和Queue，应该就可以足以应付运维中的一些问题。