python 多线程并发threading & 任务队列Queue

Posted 2020-09-17

https://docs.python.org/3.7/library/concurrency.html
python程序默认是单线程的，也就是说在前一句语句执行完之前后面的语句不能继续执行
先感受一下线程，一般情况下：

def testa():
    sleep(1)
    print "a"

def testb():
    sleep(1)
    print "b"

testa()
testb()
#先隔出一秒打印出a，再过一秒打出b

但是如果用了threading的话：

ta = threading.Thread(target=testa)
tb = threading.Thread(target=testb)
for t in [ta,tb]:
    t.start()
for t in [ta,tb]:
    t.join()
print "DONE"

#输出是ab或者ba（紧贴着的）然后空一行再来DONE的结果。

得到这样的结果是因为这样的，在start之后，ta首先开始跑，但是主线程（脚本本身）没有等其完成就继续开始下一轮循环，然后tb也开始了，在之后的一段时间里，ta和tb两条线程（分别代表了testa和testb这两个过程）共同执行。相对于一个个迭代而言，这样做无疑是大大提高了运行的速度。

　　Thread类为线程的抽象类，其构造方法的参数target指向一个函数对象，即该线程的具体操作。此外还可以有args=<tuple>来给target函数传参数。需要注意的是当传任何一个序列进去的话Thread会自动把它分解成单个单个的元素然后分解传给target函数。我估计在定义的时候肯定是*args了。

　　join方法是个很tricky的东西，至今还不是很清楚地懂这是个什么玩意儿。join([timeout])方法阻塞了主线程，直到调用此方法的子线程完成之后主线程才继续往下运行。（之前我糊里糊涂地把join就紧紧接在start后面写了，如果这么写了的话那么多线程在速度上就毫无优势，和单线程一样了= =）。而像上面这个示例一样，先一个遍历把所有线程 都启动起来，再用一个遍历把所有线程都join一遍似乎是比较通行的做法。

关于线程锁

　　多线程程序涉及到一个问题，那就是当不同线程要对同一个资源进行修改或利用时会出现混乱，所以有必要引入线程锁。

　　可以通过Thread.Lock类来创建简单的线程锁。lock = threading.Lock()即可。在某线程start之前，让lock.acquire()，且lock在acquire()之后不能再acquire，否则会报错。当线程结束后调用lock.release()来释放锁就好了。一般而言，有锁的多线程场景可以提升一部分效率，但在写文件等时机下会有阻塞等待的情况。相比之下，无所多线程场景可以进一步提升效率，但是可能会引起读写冲突等问题，所以要慎用。一定要确认各个线程间没有共同的资源之类的问题后再实行无锁多线程。

 

　　●　　以上的包装线程的方式是一种面向过程的方法，下面介绍一下如何面向对象地来抽象线程

　　面向对象地抽象线程需要自定义一个类继承Thread类。比如自定义class MyThread(Thread)。这个类的一个实例就是代表了一个线程，然后通过重载这个类中的run方法（是run，不是start！！但start的动作确实就是调用run）来执行具体的操作。此时锁可以作为一个构造方法的参数，将一个锁传进不同的实例中以实现线程锁控制。比如：

#方法二：从Thread继承，并重写run()
class MyThread(threading.Thread):
    def __init__(self,arg):
        super(MyThread, self).__init__()#注意：一定要显式的调用父类的初始化函数。
        self.arg=arg
    def run(self):#定义每个线程要运行的函数
        time.sleep(1)
        print ‘the arg is:%s\r‘ % self.arg

for i in xrange(4):
    t =MyThread(i)
    t.start()

print ‘main thread end!‘

Thread类还有以下的一些方法，自定义的类也可以调用

　　　　getName()

　　　　setName(...)　　//其实Thread类在构造方法中有一个name参数，可以为相应的线程取一个名字。这两个方法就是相关这个名字属性的

　　　　isAlive()　　一个线程从start()开始到run()结束的过程中没有异常，则其实alive的。

　　　　setDaemon(True/False)　　是否设置一个线程为守护线程。当你设置一个线程为守护线程之后，程序不会等待这个线程结束再退出程序，可参考http://blog.csdn.net/u012063703/article/details/51601579

 

　　●　　除了Thread类，threading中还有以下一些属性，简单介绍一下：

　　　　Timer类，Timer(int,target=func)　　和Thread类类似，只不过它在int秒过后才以target指定的函数开始线程运行

　　　　currentThread()　　获得当前线程对象

　　　　activeCount()　　获得当前活动的线程总个数

　　　　enumerate()　　获得所有活动线程的列表

　　　　settrace(func)　　设置一跟踪函数，在run执行前执行

　　　　setprofile(func)　　设置一跟踪函数，在run执行完毕之后执行

Queue用于建立和操作队列，常和threading类一起用来建立一个简单的线程队列。

　　首先，队列有很多种，根据进出顺序来分类，可以分成

　　　　Queue.Queue(maxsize)　　FIFO（先进先出队列）

　　　　Queue.LifoQueue(maxsize)　　LIFO（先进后出队列）

　　　　Queue.PriorityQueue(maxsize)　　为优先度越低的越先出来

　　　　如果设置的maxsize小于1，则表示队列的长度无限长

　　FIFO是常用的队列，其一些常用的方法有：

　　　　Queue.qsize()　　返回队列大小

　　　　Queue.empty()　　判断队列是否为空

　　　　Queue.full()　　判断队列是否满了

　　　　Queue.get([block[,timeout]])　　从队列头删除并返回一个item，block默认为True，表示当队列为空却去get的时候会阻塞线程，等待直到有有item出现为止来get出这个item。如果是False的话表明当队列为空你却去get的时候，会引发异常。在block为True的情况下可以再设置timeout参数。表示当队列为空，get阻塞timeout指定的秒数之后还没有get到的话就引发Full异常。

　　　　Queue.put(...[,block[,timeout]])　　向队尾插入一个item，同样若block=True的话队列满时就阻塞等待有空位出来再put，block=False时引发异常。同get的timeout，put的timeout是在block为True的时候进行超时设置的参数。

　　　　Queue.task_done()　　从场景上来说，处理完一个get出来的item之后，调用task_done将向队列发出一个信号，表示本任务已经完成

　　　　Queue.join()　　监视所有item并阻塞主线程，直到所有item都调用了task_done之后主线程才继续向下执行。这么做的好处在于，假如一个线程开始处理最后一个任务，它从任务队列中拿走最后一个任务，此时任务队列就空了但最后那个线程还没处理完。当调用了join之后，主线程就不会因为队列空了而擅自结束，而是等待最后那个线程处理完成了。

 

　　结合threading和Queue可以构建出一个简单的生产者-消费者模型，比如：

import threading  
    import Queue  
    import time  
    class worker(threading.Thread):  
     def __init__(self,queue):  
      threading.Thread.__init__(self)  
      self.queue=queue  
      self.thread_stop=False  
       
     def run(self):  
      while not self.thread_stop:  
       print("thread%d %s: waiting for tast" %(self.ident,self.name))  
       try:  
        task=q.get(block=True, timeout=20)#接收消息  
       except Queue.Empty:  
        print("Nothing to do!i will go home!")  
        self.thread_stop=True  
        break  
       print("task recv:%s ,task No:%d" % (task[0],task[1]))  
       print("i am working")  
       time.sleep(3)  
       print("work finished!")  
       q.task_done()#完成一个任务  
       res=q.qsize()#判断消息队列大小  
       if res>0:  
        print("fuck!There are still %d tasks to do" % (res))  
       
     def stop(self):  
      self.thread_stop = True  
       
    if __name__ == "__main__":  
     q=Queue.Queue(3)  
     worker=worker(q)  
     worker.start()  
     q.put(["produce one cup!",1], block=True, timeout=None)#产生任务消息  
     q.put(["produce one desk!",2], block=True, timeout=None)  
     q.put(["produce one apple!",3], block=True, timeout=None)  
     q.put(["produce one banana!",4], block=True, timeout=None)  
     q.put(["produce one bag!",5], block=True, timeout=None)  
     print("***************leader:wait for finish!")  
     q.join()#等待所有任务完成  
     print("***************leader:all task finished!")  

输出是这样的

thread139958685849344 Thread-1: waiting for tast 1
    task recv:produce one cup! ,task No:1
    i am working
    work finished!
    fuck!There are still 3 tasks to do
    thread139958685849344 Thread-1: waiting for tast 1
    task recv:produce one desk! ,task No:2
    i am workingleader:wait for finish!
    work finished!
    fuck!There are still 3 tasks to do
    thread139958685849344 Thread-1: waiting for tast 1
    task recv:produce one apple! ,task No:3
    i am working
    work finished!
    fuck!There are still 2 tasks to do
    thread139958685849344 Thread-1: waiting for tast 1
    task recv:produce one banana! ,task No:4
    i am working
    work finished!
    fuck!There are still 1 tasks to do
    thread139958685849344 Thread-1: waiting for tast 1
    task recv:produce one bag! ,task No:5
    i am working
    work finished!
    thread139958685849344 Thread-1: waiting for tast 1
     ***************leader:all task finished!
    Nothing to do!i will go home!

上例中并没有性能的提升（毕竟还是只有一个线程在跑）。线程队列的意义并不是进一步提高运行效率，而是使线程的并发更加有组织。可以看到，在增加了线程队列之后，程序对于线程的并发数量就有了控制。新线程想要加入队列开始执行，必须等一个既存的线程完成之后才可以。举个例子，比如

for i in range(x):
  t = MyThread(queue)
  t.start()

x在这里是个变量，我们不知道这个循环会触发多少线程并发，如果多的话就会很冒险。但是有了队列之后，把一个队列作为所有线程构建线程对象时的一个参数，让线程必须按照这个队列规定的大小来执行的话，就不担心过多线程带来的危险了。