Python菜鸟之路：Python基础-线程、进程、协程

　　上节内容，简单的介绍了线程和进程，并且介绍了Python中的GIL机制。本节详细介绍线程、进程以及协程的概念及实现。

线程

基本使用

方法1： 创建一个threading.Thread对象，在它的初始化函数（__init__）中将可调用对象作为参数传入

import threading
import time

def worker():
    time.sleep()
    print("test")

for i in range():
    t = threading.Thread(target=worker)
    t.start()

线程创建方法1

方法2：通过继承Thread类，重写run方法

import threading

class MyThread(threading.Thread):
    def __init__(self, func, args):
        self.func = func
        self.args = args
        super(MyThread, self).__init__()

    def run(self):
        self.func(self.args)

def f2(arg):
    print(arg)

obj = MyThread(f2, 123)
obj.start()

线程创建方法2

线程锁（互斥锁，信号量，条件变量，事件）

　　上边的2个创建多线程的例子中，多个线程之间没有任何关系。假设这样一种情况，有一个全局的计数num，每个线程获取这个全局的计数，根据num进行一些处理，然后将num加1。很容易写出这样的代码：

import threading
import time

NUM = 0
class MyThread(threading.Thread):
    def run(self):
        global NUM
        NUM += 1
        time.sleep(0.5)
        msg = self.name+' set num to '+str(NUM)
        print(msg)
if __name__ == '__main__':
    for i in range(5):
        t = MyThread()
        t.start()

out:
Thread-5 set num to 2
Thread-3 set num to 3
Thread-2 set num to 5
Thread-1 set num to 5
Thread-4 set num to 4

　　问题产生的原因就是没有控制多个线程对同一资源的访问，对数据造成破坏，使得线程运行的结果不可预期。这种现象称为“线程不安全”。因此有了线程锁的概念。

　　线程同步能够保证多个线程安全访问竞争资源，最简单的同步机制是引入互斥锁。互斥锁为资源引入一个状态：锁定/非锁定。某个线程要更改共享数据时，先将其锁定，此时资源的状态为“锁定”，其他线程不能更改；直到该线程释放资源，将资源的状态变成“非锁定”，其他的线程才能再次锁定该资源。互斥锁保证了每次只有一个线程进行写入操作，从而保证了多线程情况下数据的正确性。

　　引入锁的话，上面的代码可以写作

import threading
import time

NUM = 0
class MyThread(threading.Thread):
    def run(self, l):
        l.acquire()
        global NUM
        NUM += 1
        time.sleep(0.5)
        msg = self.name+' set num to '+str(NUM)
        l.release()
        print(msg)
if __name__ == '__main__':
    lock = threading.Lock()
    for i in range(5):
        t = MyThread()
        t.run(lock)

out:
Thread-1 set num to 1
Thread-2 set num to 2
Thread-3 set num to 3
Thread-4 set num to 4
Thread-5 set num to 5

加线程锁

　　可以看出，加了线程锁之后，尝试任何次数的执行，最终结果都是一样的，消除了线程不安全的隐患。过程是这样的：我们先建立了一个threading.Lock类对象lock,在run方法里，我们使用lock.acquire()获得了这个锁。此时，其他的线程就无法再获得该锁了，他们就会阻塞在“l.acquire()”这里，直到锁被另一个线程释放l.release()。

　　上边的锁机制，叫做“互斥锁”。同一时间仅允许一个线程进行更改数据。而信号量(Semaphore)是同时允许一定数量的线程更改数据 ，比如厕所有3个坑，那最多只允许3个人上厕所，后面的人只能等里面有人出来了才能再进去。

信号量

import threading,time

def run(n):
    semaphore.acquire()
    time.sleep(1)
    print("run the thread: %s" %n)
    semaphore.release()

if __name__ == '__main__':

    num= 0
    semaphore  = threading.BoundedSemaphore(5) #最多允许5个线程同时运行
    for i in range(20):
        t = threading.Thread(target=run,args=(i,))
        t.start()

　　执行之后会发现，每个1秒，会有5个线程执行，有print的操作。

死锁

　　如果多个线程要调用多个现象，而A线程调用A锁占用了A对象，B线程调用了B锁占用了B对象,A线程不能调用B对象，B线程不能调用A对象，于是一直等待。这就造成了线程“死锁”。
　　Threading模块中，也有一个类，RLock，称之为可重入锁。该锁对象内部维护着一个Lock和一个counter对象。counter对象记录了acquire的次数，使得资源可以被多次require。最后，当所有RLock被release后，其他线程才能获取资源。在同一个线程中，RLock.acquire可以被多次调用，利用该特性，可以解决部分死锁问题。

class _RLock:
    """This class implements reentrant lock objects.

    A reentrant lock must be released by the thread that acquired it. Once a
    thread has acquired a reentrant lock, the same thread may acquire it
    again without blocking; the thread must release it once for each time it
    has acquired it.

    """

    def __init__(self):
        self._block = _allocate_lock()
        self._owner = None
        # 初始化一个计数器
        self._count = 0

    def __repr__(self):
        owner = self._owner
        try:
            owner = _active[owner].name
        except KeyError:
            pass
        return "<%s %s.%s object owner=%r count=%d at %s>" % (
            "locked" if self._block.locked() else "unlocked",
            self.__class__.__module__,
            self.__class__.__qualname__,
            owner,
            self._count,
            hex(id(self))
        )

    def acquire(self, blocking=True, timeout=-1):
        """Acquire a lock, blocking or non-blocking.

        When invoked without arguments: if this thread already owns the lock,
        increment the recursion level by one, and return immediately. Otherwise,
        if another thread owns the lock, block until the lock is unlocked. Once
        the lock is unlocked (not owned by any thread), then grab ownership, set
        the recursion level to one, and return. If more than one thread is
        blocked waiting until the lock is unlocked, only one at a time will be
        able to grab ownership of the lock. There is no return value in this
        case.

        When invoked with the blocking argument set to true, do the same thing
        as when called without arguments, and return true.

        When invoked with the blocking argument set to false, do not block. If a
        call without an argument would block, return false immediately;
        otherwise, do the same thing as when called without arguments, and
        return true.

        When invoked with the floating-point timeout argument set to a positive
        value, block for at most the number of seconds specified by timeout
        and as long as the lock cannot be acquired.  Return true if the lock has
        been acquired, false if the timeout has elapsed.

        """
        me = get_ident()
        if self._owner == me:
            # 每次调用acquire，计数器加1
            self._count += 1
            return 1
        rc = self._block.acquire(blocking, timeout)
        if rc:
            self._owner = me
            self._count = 1
        return rc

    __enter__ = acquire

    def release(self):
        """Release a lock, decrementing the recursion level.

        If after the decrement it is zero, reset the lock to unlocked (not owned
        by any thread), and if any other threads are blocked waiting for the
        lock to become unlocked, allow exactly one of them to proceed. If after
        the decrement the recursion level is still nonzero, the lock remains
        locked and owned by the calling thread.

        Only call this method when the calling thread owns the lock. A
        RuntimeError is raised if this method is called when the lock is
        unlocked.

        There is no return value.

        """
        if self._owner != get_ident():
            raise RuntimeError("cannot release un-acquired lock")
        # 每次调用release,计数器减1
        self._count = count = self._count - 1
        if not count:
            self._owner = None
            self._block.release()

    def __exit__(self, t, v, tb):
        self.release()

    # Internal methods used by condition variables

    def _acquire_restore(self, state):
        self._block.acquire()
        self._count, self._owner = state

    def _release_save(self):
        if self._count == 0:
            raise RuntimeError("cannot release un-acquired lock")
        count = self._count
        self._count = 0
        owner = self._owner
        self._owner = None
        self._block.release()
        return (count, owner)

    def _is_owned(self):
        return self._owner == get_ident()

python3中Rlock的实现代码

　　Ps:在使用RLock的时候，要注意RLock.acquire和Rlock.release必须是成对出现的。否则会造成资源死锁，其他线程一直等待的情况，即“死锁”。

死锁发生的条件：

• 互斥条件：线程对资源的访问是排他性的，如果一个线程对占用了某资源，那么其他线程必须处于等待状态，直到资源被释放。

• 请求和保持条件：线程T1至少已经保持了一个资源R1占用,但又提出对另一个资源R2请求，而此时，资源R2被其他线程T2占用，于是该线程T1也必须等待，但又对自己保持的资源R1不释放。

• 不剥夺条件：线程已获得的资源，在未使用完之前，不能被其他线程剥夺，只能在使用完以后由自己释放。

• 环路等待条件：在死锁发生时，必然存在一个“进程-资源环形链”，即：{p0,p1,p2,...pn},进程p0（或线程）等待p1占用的资源，p1等待p2占用的资源，pn等待p0占用的资源。（最直观的理解是，p0等待p1占用的资源，而p1而在等待p0占用的资源，于是两个进程就相互等待）

上面的文字太生硬，看一个比喻就懂了:

　　迎面开来的汽车A和汽车B过马路，汽车A得到了半条路的资源（满足死锁发生条件1：资源访问是排他性的，我占了路你就不能上来，除非你爬我头上去），汽车B占了汽车A的另外半条路的资源，A想过去必须请求另一半被B占用的道路（死锁发生条件2：必须整条车身的空间才能开过去，我已经占了一半，尼玛另一半的路被B占用了），B若想过去也必须等待A让路，A是辆兰博基尼，B是开奇瑞QQ的屌丝，A素质比较低开窗对B狂骂：快给老子让开，B很生气，你妈逼的，老子就不让（死锁发生条件3：在未使用完资源前，不能被其他线程剥夺），于是两者相互僵持一个都走不了（死锁发生条件4：环路等待条件），而且导致整条道上的后续车辆也走不了。

其他锁：活锁

　　活锁：是指线程1可以使用资源，但它很礼貌，让其他线程先使用资源，线程2也可以使用资源，但它很绅士，也让其他线程先使用资源。这样你让我，我让你，最后两个线程都无法使用资源。

一个比喻：马路中间有条小桥，只能容纳一辆车经过，桥两头开来两辆车A和B，A比较礼貌，示意B先过，B也比较礼貌，示意A先过，结果两人一直谦让谁也过不去。

其他锁：饥饿

　　是指如果线程T1占用了资源R，线程T2又请求*R，于是T2等待。T3也请求资源R，当T1释放了R上的*后，系统首先批准了T3的请求，T2仍然等待。然后T4又请求*R，当T3释放了R上的*之后，系统又批准了T4的请求......，T2可能永远等待。

一个比喻：在“首堵”北京的某一天，天气阴沉，空气中充斥着雾霾和地沟油的味道，某个苦逼的临时工交警正在处理塞车，有两条道A和B上都堵满了车辆，其中A道堵的时间最长，B相对相对堵的时间较短，这时，前面道路已疏通，交警按照最佳分配原则，示意B道上车辆先过，B道路上过了一辆又一辆，A道上排队时间最长的确没法通过，只能等B道上没有车辆通过的时候再等交警发指令让A道依次通过，这也就是ReentrantLock显示锁里提供的不公平锁机制（当然了，ReentrantLock也提供了公平锁的机制，由用户根据具体的使用场景而决定到底使用哪种锁策略），不公平锁能够提高吞吐量但不可避免的会造成某些线程的饥饿。

补充概念：同步阻塞

　　当一个线程调用锁的acquire()方法获得锁时，锁就进入“locked”状态。每次只有一个线程可以获得锁。如果此时另一个线程试图获得这个锁，该线程就会变为“blocked”状态，称为“同步阻塞”

队列

　　Python中的队列，常用的包括以下四种：1.先进先出队列  2.后进先出队列. 3.优先级队列  4.双向队列。这四种队列的使用，在queue模块中已经定义好了，具体使用如下↓

1）先进先出队列

import queue

# 定义队列最大长度
q = queue.Queue(maxsize=5)
# q.empty() 判断队列是否为空
print('if q is empty: ', q.empty())
# 向队列中push元素
q.put(1)
q.put(2)
q.put(3)
q.put(4)
# 判断队列长度是否达到上限
print('if q is full:', q.full())
print('queue size :' , q.qsize())
q.put(5)
print('if q is full:', q.full())
# 向队列中put数据，如果队列满，则立即抛出异常，不阻塞
q.put_nowait(6)
# 向队列中put数据,默认为block为True,表示阻塞，timeout表示最大等待时间，如果这段时间内put失败，则抛出异常
q.put(7, block=True, timeout=2)
print('if q is empty: ', q.empty())
# 返回当前队列元素个数
print('queue size :' , q.qsize())
# 从队列中获取单个数据
print(q.get())
print(q.get())
print(q.get())
print(q.get())
print(q.get())
# q.get(block, timeout) 默认block为True,表示是否阻塞。timeout表示阻塞最长时间，如果这段时间内无法获取元素，则抛出异常
print(q.get(block=True, timeout=2))
# block=False时，如果get异常，立即抛出错误
print(q.get(block=False))
# 从队列中获取元素，如果获取异常则抛出异常，不阻塞。
# print(q.get_nowait())
print('if q is empty: ', q.empty())

queue.Queue()用法

　　q.task_done()  # 每次从queue中get一个数据之后，当处理好相关问题，最后调用该方法，以提示q.join()是否停止阻塞，让线程向前执行或者退出；

　　q.join()            # 阻塞，直到queue中的数据均被删除或者处理。为队列中的每一项都调用一次。通常用在consumer中。例子如下：

import queue

q = queue.Queue()

q.put(1)
q.put(1)
q.put(1)

print('get item: ', q.get())
q.task_done()
print('get item: ', q.get())
# q.task_done()
print('get item: ', q.get())
# q.task_done()
q.join()

out:
get item:  1
get item:  1
get item:  1
# ...当task_done的调用次数小于get的次数时，会一直阻塞

task_done和Join

from queue import Queue
from threading import Thread

class producer(Thread):
    def __init__(self ,q):
        super().__init__()
        self.q = q

    def run(self):
        self.count=5
        while self.count>0:
            if self.count==1:
               self.count -=1
               self.q.put(2)
            else:
               self.count -=1
               self.q.put(1)

class consumer(Thread):

    def __init__(self,q):
        super().__init__()
        self.q = q

    def run(self):
        while True:
            data = self.q.get()
            if data==2:
                print("stop because data=",data)
                self.q.task_done()
                break
            else:
                print("data is good,data=",data)
                self.q.task_done()

def main():
    qq = Queue()
    p = producer(qq)
    c = consumer(qq)
    p.setDaemon(True)
    c.setDaemon(True)
    p.start()
    c.start()
    qq.join()
    print("queue is complete")

if __name__ == '__main__':
    main()

task_done和join实践案例

2）后进先出队列

import queue

q = queue.LifoQueue()
q.put(1)
q.put(2)
q.put(3)
print('first get: ', q.get())
print('second get: ', q.get())
print('third get: ', q.get())

out:
first get:  3
second get:  2
third get:  1

queue.LifoQueue后进先出队列

3）优先级队列

import queue

q1 = queue.PriorityQueue() # 如果优先级相同，谁先放进去，先取出谁
q1.put((1, 'alex1'))
q1.put((2, 'alex2'))
q1.put((1, 'alex3'))

print(q1.get())
print(q1.get())
print(q1.get())

out:
(1, 'alex1')
(1, 'alex3')
(2, 'alex2')

queue.PriorityQueue()优先级队列

　　Ps:优先级相同，谁先存，就先取出谁。

4）双向队列　　

import queue

q2 = queue.deque()
q2.append(1)
q2.append(2) # 从右侧增加数据
q2.append(3)
q2.appendleft(4) # 从左侧增加数据

q2.extend([5,6]) # 从右侧增加一个可迭代对象数据
q2.extendleft([7]) # 从左侧增加一个可迭代对象数据

print(q2)
q2.pop()  # 从右侧取数据
q2.popleft() # 从左侧取数据
print(q2)

out:
deque([7, 4, 1, 2, 3, 5, 6])
deque([4, 1, 2, 3, 5])

queue.deque()双向队列

生产者、消费者模型

　　有一个或多个生产者生产某种类型的数据，并放置在缓冲区（可以是数组也可以是队列等数据结构）中；有一个消费者可以从缓冲区中取数据，每次取一项；系统保证避免对缓冲区的重复操作，也就是说在任何时候只有一个主体（生产者或消费者）可以访问缓冲区。问题要确保缓冲区不溢出，即当缓冲区满时，生成者不会继续向其中添加数据；当缓冲区空时，消费者不会从中移走数据。该模型可以解决：阻塞问题和程序解耦。下面是一个吃包子和做包子的案例代码：

import queue
import threading
import time

q = queue.Queue(20)

def productor(arg):
    while True:
        q.put(str(arg) + '号厨师弄出来的包子')

def consumer(arg):
    while True:
        print( arg, q.get())
        time.sleep(2)

for i in range(3):
    t = threading.Thread(target=productor, args=(i,))
    t.start()

for j in range(20):
    t = threading.Thread(target=consumer, args=(j,))
    t.start()

生产者、消费者模型案例代码

　　生产者的工作是产生一块数据，放到buffer中，如此循环。与此同时，消费者在消耗这些数据（例如从buffer中把它们移除），每次一块。这里的关键词是“同时”。所以生产者和消费者是并发运行的，我们需要对生产者和消费者做线程分离。假设这样一种情况，生产者负责往队列中存数据，消费者负责从队列中取数据。这两个线程同时运行，会存在这样一种情况：在某一时间点，消费者把所有东西消耗完毕而生产者还在挂起（sleep）。消费者尝试继续进行消耗，但此时队列为空，出现异常。我们把这个实现作为错误行为（wrong behavior）。测试代码如下：

from threading import Thread, Lock
import time
import random

queue = []
lock = Lock()

class ProducerThread(Thread):
  def run(self):
    nums = range(5) #Will create the list [0, 1, 2, 3, 4]
    global queue
    while True:
      num = random.choice(nums) #Selects a random number from list [0, 1, 2, 3, 4]
      lock.acquire()
      queue.append(num)
      print("Produced", num)
      lock.release()
      time.sleep(random.random())

class ConsumerThread(Thread):
  def run(self):
    global queue
    while True:
      lock.acquire()
      if not queue:
        print("Nothing in queue, but consumer will try to consume")
        exit()
      num = queue.pop(0)
      print("Consumed", num)
      lock.release()
      time.sleep(random.random())

ProducerThread().start()
ConsumerThread().start()

out:
Produced 1
Consumed 1
Produced 0
Produced 4
Produced 1
Consumed 0
Consumed 4
Produced 2
Consumed 1
Consumed 2
Produced 3
Consumed 3
Nothing in queue, but consumer will try to consume

错误行为代码

　　那么，正确行为应该是怎样的呢？

　　当队列中没有任何数据的时候，消费者应该停止运行并等待(wait），而不是继续尝试进行消耗。而当生产者在队列中加入数据之后，应该有一个渠道去告诉（notify）消费者。然后消费者可以再次从队列中进行消耗，而IndexError不再出现。而条件（Condition）正是用来解决这一问题。

条件变量

　　互斥锁是最简单的线程同步机制，Python提供的Condition对象提供了对复杂线程同步问题的支持。Condition被称为条件变量，除了提供与Lock类似的acquire和release方法外，还提供了wait和notify方法。线程首先acquire一个条件变量，然后判断一些条件。如果条件不满足则wait；如果条件满足，进行一些处理改变条件后，通过notify方法通知其他线程，其他处于wait状态的线程接到通知后会重新判断条件。不断的重复这一过程，从而解决复杂的同步问题。

　　可以认为Condition对象维护了一个锁（Lock/RLock)和一个waiting池。线程通过acquire获得Condition对象，当调用wait方法时，线程会释放Condition内部的锁并进入blocked状态，同时在waiting池中记录这个线程。当调用notify方法时，Condition对象会从waiting池中挑选一个线程，通知其调用acquire方法尝试取到锁。

　　Condition对象的构造函数可以接受一个Lock/RLock对象作为参数，如果没有指定，则Condition对象会在内部自行创建一个RLock。

除了notify方法外，Condition对象还提供了notifyAll方法，可以通知waiting池中的所有线程尝试acquire内部锁。由于上述机制，处于waiting状态的线程只能通过notify方法唤醒，所以notifyAll的作用在于防止有线程永远处于沉默状态。

　　假设有这样一种场景：有一群生产者(Producer)和一群消费者（Consumer）通过一个市场来交互产品。生产者的”策略“是如果市场上剩余的产品少于1000个，那么就生产100个产品放到市场上；而消费者的”策略“是如果市场上剩余产品的数量多余100个，那么就消费3个产品。用Condition解决生产者与消费者问题的代码如下：

　　参数说明: con.acquire和con.wait   通常在加锁的时候配合使用,表示在这里等待　　　　　

　　　　　　　con.acquire() # 这三个必须放一起，是固定用法，通常用于解锁的时候
　　　　　　　con.notify(num) # 表示一次放多少个线程过去
　　　　　　　con.release()

第一种方式：

import threading
import time

class Producer(threading.Thread):
    def run(self):
        global count
        while True:
            if con.acquire():
                if count > 1000:
                    con.wait()
                else:
                    count = count+100
                    msg = self.name+' produce 100, count=' + str(count)
                    print(msg)
                    con.notify()
                con.release()
                time.sleep(1)

class Consumer(threading.Thread):
    def run(self):
        global count
        while True:
            if con.acquire():
                if count < 100:
                    con.wait()
                else:
                    count = count-3
                    msg = self.name+' consume 3, count='+str(count)
                    print(msg)
                    con.notify()
                con.release()
                time.sleep(1)

count = 500
con = threading.Condition()

def test():
    for i in range(2):
        p = Producer()
        p.start()
    for i in range(5):
        c = Consumer()
        c.start()
if __name__ == '__main__':
    test()

out:
Thread-1 produce 100, count=600
Thread-2 produce 100, count=700
Thread-3 consume 3, count=697
Thread-4 consume 3, count=694
Thread-5 consume 3, count=691
Thread-6 consume 3, count=688
Thread-7 consume 3, count=685
Thread-2 produce 100, count=785
Thread-1 produce 100, count=885
Thread-4 consume 3, count=882
Thread-3 consume 3, count=879
Thread-5 consume 3, count=876
Thread-6 consume 3, count=873
Thread-7 consume 3, count=870
Thread-1 produce 100, count=970
Thread-2 produce 100, count=1070
Thread-4 consume 3, count=1067
Thread-3 consume 3, count=1064
Thread-6 consume 3, count=1061
Thread-7 consume 3, count=1058
Thread-5 consume 3, count=1055
Thread-4 consume 3, count=1052
Thread-3 consume 3, count=1049
Thread-7 consume 3, count=1046
Thread-6 consume 3, count=1043
Thread-5 consume 3, count=1040
Thread-4 consume 3, count=1037
Thread-7 consume 3, count=1034
Thread-3 consume 3, count=1031
Thread-5 consume 3, count=1028
Thread-6 consume 3, count=1025
Thread-5 consume 3, count=1022
Thread-6 consume 3, count=1019
Thread-4 consume 3, count=1016
Thread-7 consume 3, count=1013
Thread-3 consume 3, count=1010
Thread-5 consume 3, count=1007
Thread-6 consume 3, count=1004
Thread-4 consume 3, count=1001
Thread-7 consume 3, count=998
Thread-3 consume 3, count=995
Thread-2 produce 100, count=1095
Thread-5 consume 3, count=1092
Thread-6 consume 3, count=1089
Thread-4 consume 3, count=1086
Thread-7 consume 3, count=1083
Thread-3 consume 3, count=1080
Thread-5 consume 3, count=1077
Thread-6 consume 3, count=1074
Thread-3 consume 3, count=1071
Thread-4 consume 3, count=1068
Thread-7 consume 3, count=1065
Thread-5 consume 3, count=1062
Thread-6 consume 3, count=1059
Thread-4 consume 3, count=1056
Thread-7 consume 3, count=1053
Thread-3 consume 3, count=1050
Thread-5 consume 3, count=1047
Thread-6 consume 3, count=1044
Thread-7 consume 3, count=1041
Thread-3 consume 3, count=1038
Thread-4 consume 3, count=1035

condition解决同步问题案例代码

　第二种方式：con.wait_for() 表示等待某个条件成立，如果成立的话继续，不成立的话等待，接收True或者False

　　　　　　　con.acquire() # 这三个同样一起使用
　　　　　　　con.wait_for(条件) # 等待条件返回True
　　　　　　　con.release()

　

import threading

def condition():
    ret = False
    r = input(">>> ")
    if r == 'true':
        ret = True
    else:
        pass
    return ret

def func(i, con):
    print(i)
    con.acquire()  # con.acquire和con.wait 必须配合使用，表示在这里等待
    con.wait_for(condition)
    print(i+100)
    con.release()

c = threading.Condition()
for i in range(10):
    t = threading.Thread(target=func, args=(i, c,))
    t.start()

condition方式2

　小结：目前，可以解决线程间“同步问题”的方法，已经有2种：互斥锁和条件变量。其实队列也是一种！！！

import threading
import time
from queue import Queue

class Producer(threading.Thread):
    def run(self):
        global queue
        count = 0
        while True:
            for i in range(100):
                if queue.qsize() > 1000:
                     pass
                else:
                     count = count +1
                     msg = '生成产品'+str(count)
                     queue.put(msg)
                     print(msg)
            time.sleep(1)

class Consumer(threading.Thread):
    def run(self):
        global queue
        while True:
            for i in range(3):
                if queue.qsize() < 100:
                    pass
                else:
                    msg = self.name + '消费了 '+queue.get()
                    print(msg)
            time.sleep(1)

queue = Queue()

def test():
    for i in range(500):
        queue.put('初始产品'+str(i))
    for i in range(2):
        p = Producer()
        p.start()
    for i in range(5):
        c = Consumer()
        c.start()
if __name__ == '__main__':
    test()

队列解决同步问题代码案例

　　那么除了条件，还有其他方法来解决生产者与消费者模型中的“同步问题”么？事件（event）

事件

　　很多时候，线程之间会有互相通信的需要。常见的情形是次要线程为主要线程执行特定的任务，在执行过程中需要不断报告执行的进度情况。前面的条件变量同步已经涉及到了线程间的通信（threading.Condition的notify方法）。更通用的方式是使用threading.Event对象。注意：事件其实也是一种锁，只不过这把锁是把线程全部锁住，直到信号被set为True的时候，线程才恢复运行。

　　python线程的事件用于主线程控制其他线程的执行，事件主要提供了三个方法 set、wait、clear。

　　threading.Event可以使一个线程等待其他线程的通知。其内置了一个标志，初始值为False。线程通过wait()方法进入等待状态，直到另一个线程调用set()方法将内置标志设置为True时，Event通知所有等待状态的线程恢复运行。还可以通过isSet()方法查询Envent对象内置状态的当前值。代码如下：

import threading

def func(i, e):
    print(i)
    e.wait() # 表示在这里检测信号。如果检测到为红灯，则停止。如果为绿灯，则放行
    print(i+100)

event = threading.Event()

for i in range(10):
    t = threading.Thread(target=func, args=(i,event))
    t.start()

event.clear() # 全部都停止，设置为红灯

inp = input(">>> ")
if inp == "":
    event.set()  # 表示放行，设置为绿灯

threading.Event()时间代码案例

扩展：自定义线程池

import queue
import threading, time

class ThreadPool:
    def __init__(self, maxsize=5):
        self.maxsize = maxsize
        self._q = queue.Queue(maxsize)
        for i in range(maxsize):
            self._q.put(threading.Thread)

    def get_thread(self):
        return self._q.get()

    def add_thread(self):
        self._q.put(threading.Thread)

pool = ThreadPool(5)
def task(arg, p):
    print(arg)
    time.sleep(1)
    p.add_thread()

for i in range(100):
    t = pool.get_thread() # t是一个类
    obj = t(target=task, args=(i, pool,))
    obj.start()

low版本的线程池，缺点：1. 线程池中线程不能被重用，一直被创建和取出 2. 线程池长度及所占内存资源有浪费，可能过多(如果任务数小于线程池大小)。如果线程执行比较快，可能只需要少量的线程就能完成

import queue
import threading
import contextlib
import time

StopEvent = object()

class ThreadPool(object):

    def __init__(self, max_num, max_task_num = None):
        if max_task_num:
            self.q = queue.Queue(max_task_num)
        else:
            self.q = queue.Queue()
        self.max_num = max_num
        self.cancel = False
        self.terminal = False
        self.generate_list = []
        # 当前空闲多少线程
        self.free_list = []

    def run(self, func, args, callback=None):
        """
        线程池执行一个任务
        :param func: 任务函数
        :param args: 任务函    数所需参数
        :param callback: 任务执行失败或成功后执行的回调函数，回调函数有两个参数1、任务函数执行状态；2、任务函数返回值（默认为None，即：不执行回调函数）
        :return: 如果线程池已经终止，则返回True否则None
        """
        if self.cancel:
            return
        # 如果没有空闲线程， 并且 已创建的线程小于最大线程，则创建线程
        if len(self.free_list) == 0 and len(self.generate_list) < self.max_num:
            self.generate_thread()

        # 把任务封装成一个元组，然后放到队列里
        w = (func, args, callback,)
        self.q.put(w)

    def generate_thread(self):
        """
        创建一个线程
        """
        t = threading.Thread(target=self.call)
        t.start()

    def call(self):
        """
        循环去获取任务函数并执行任务函数
        """
        current_thread = threading.currentThread
        self.generate_list.append(current_thread)

        # 取任务，任务是元组的形式
        event = self.q.get()
        while event != StopEvent:

            func, arguments, callback = event
            try:
                # 执行函数
                result = func(*arguments)
                success = True
            except Exception as e:
                success = False
                result = None

            if callback is not None:
                try:
                    callback(success, result)
                except Exception as e:
                    pass

            # 如果执行完函数之后，设置为空闲
            with self.worker_state(self.free_list, current_thread):
                if self.terminal:
                    event = StopEvent
                else:
                    event = self.q.get()
        else:
            # 移除当前线程
            self.generate_list.remove(current_thread)

    def close(self):
        """
        执行完所有的任务后，所有线程停止
        """
        self.cancel = True
        full_size = len(self.generate_list)
        while full_size:
            self.q.put(StopEvent)
            full_size -= 1

    def terminate(self):
        """
        无论是否还有任务，终止线程
        """
        self.terminal = True

        while self.generate_list:
            self.q.put(StopEvent)

        self.q.empty()

    @contextlib.contextmanager
    def worker_state(self, state_list, worker_thread):
        """
        用于记录线程中正在等待的线程数
        """
        state_list.append(worker_thread)
        try:
            yield
        finally:
            state_list.remove(worker_thread)

# How to use

pool = ThreadPool(5)

def callback(status, result):
    # status, execute action status
    # result, execute action return value
    pass

def action(i):
    print(i)

for i in range(30):
    ret = pool.run(action, (i,), callback)

time.sleep(5)
print(len(pool.generate_list), len(pool.free_list))
print(len(pool.generate_list), len(pool.free_list))
pool.close()
pool.terminate()

完善的线程池

进程

　　python中的多线程其实并不是真正的多线程，如果想要充分地使用多核CPU的资源，在python中大部分情况需要使用多进程。Python提供了非常好用的多进程包multiprocessing，只需要定义一个函数，Python会完成其他所有事情。借助这个包，可以轻松完成从单进程到并发执行的转换。multiprocessing支持子进程、通信和共享数据、执行不同形式的同步，提供了Process、Queue、Pipe、Lock等组件。

基本使用

　　多进程的使用方法与多线程基本一模一样。只需要将对应的模块和方法，替换为multiprocessing.Process即可。这里不再做过多的说明和举例，线程里已经很具体了。 　

from multiprocessing import Process

def foo(arg):
    print('say hi',arg)

if __name__ == "__main__":

    for i in range(10):
        p = Process(target=foo,args=(i,))
        #p.daemon = True  # 等同于线程的threading.Thread.setDaemon
        p.start()
        #p.join()

out:
say hi 0
say hi 2
say hi 4
say hi 1
say hi 3
say hi 6
say hi 8
say hi 7
say hi 5
say hi 9

多进程的基本使用

进程锁

　　当多个进程需要访问共享资源的时候，Lock可以用来避免访问的冲突。

　　同样的，进程锁也包含了其他类型，包括RLock,Semaphore（用来控制对共享资源的访问数量，例如池的最大连接数）。Event。同线程也是一样的。所属模块不同而已：multiprocessing.Lock()，multiprocessing.RLock()，multiprocessing.Semaphore(n) ，multiprocessing.Event()。具体使用案例可以查看http://www.cnblogs.com/kaituorensheng/p/4445418.html  或者将上边讲到的线程案例进行修改。

进程间数据共享

　　默认情况下，进程间的数据是不可以进行共享的。但是可以通过以下三个方法进行数据共享：queues, Array, Manager.dict()

1）queues

from multiprocessing import Process
from multiprocessing import queues
import multiprocessing

def foo(i,arg):
    arg.put(i)
    print('say hi',i,arg.qsize())

if __name__ == "__main__":
    li = queues.Queue(20,ctx=multiprocessing) # 源码中执行了ctx.lock
    for i in range(10):
        p = Process(target=foo,args=(i,li,))
        p.start()

out:
say hi 0 1
say hi 2 2
say hi 1 4
say hi 3 4
say hi 6 7
say hi 5 7
say hi 4 7
say hi 7 8
say hi 9 9
say hi 8 10

multiprocessing.queues

2）Array（不常用）

　　数组：Array的特点：1.它的内存地址是连续的，而列表不是  2. 数组中元素的数据类型，在创建的时候就已经定义好了。3. 个数一定。在创建的时候就需要指定

from multiprocessing import Process
from multiprocessing import Array

def foo(i,arg):
    arg[i] = i + 100
    for item in arg:
        print(item)
    print('================')

if __name__ == "__main__":
    li = Array('i', 5) # 创建一个最大包含5个元素的数组
    for i in range(5):
        p = Process(target=foo,args=(i,li,))
        p.start()

out:
100
0
0
0
0
================
100
101
0
0
0
================
100
101
102
0
0
================
100
101
102
103
0
================
100
101
102
103
104
================

multiprocessing.Array

　　代码中的"i"类型是包含在C中类型对应表中的：

    'c': ctypes.c_char,  'u': ctypes.c_wchar,
    'b': ctypes.c_byte,  'B': ctypes.c_ubyte,
    'h': ctypes.c_short, 'H': ctypes.c_ushort,
    'i': ctypes.c_int,   'I': ctypes.c_uint,
    'l': ctypes.c_long,  'L': ctypes.c_ulong,
    'f': ctypes.c_float, 'd': ctypes.c_double

类型对应表

3）Manager.dict（常用）

from multiprocessing import Process
from multiprocessing import Manager

def foo(i,arg):
    arg[i] = i + 100
    print(arg.values())

if __name__ == "__main__":
    obj = Manager()
    li = obj.dict()
    for i in range(5):
        p = Process(target=foo,args=(i,li,))
        #p.daemon = True
        p.start()
        p.join()
out:
[100]
[100, 101]
[100, 101, 102]
[100, 101, 102, 103]
[100, 101, 102, 103, 104]

Manager.dict实现数据共享

进程池

　　进程池在multiprocessing模块中已经定义好，只需要直接使用即可。主要包含2个方法apply和apply_async

进程池的使用

from multiprocessing import Pool
import time
def f1(arg):
    print(arg,'b')
    time.sleep(5)
    print(arg,'a')

if __name__ == "__main__":
    pool = Pool(5)
    # 定义30个任务
    for i in range(30):
        # pool.apply(func=f1,args=(i,))
        pool.apply_async(func=f1,args=(i,))

    # pool.close() # 等待所有的任务执行完毕后，修改进程状态为close。否则会阻塞到这里
    time.sleep(2)
    pool.terminate() # 立即终止全部子进程
    pool.join()  # 主进程在这里等待子进程全部执行完毕

out:
0 b
1 b
2 b
3 b
4 b  # 执行到一半已经被强制终止

Process finished with exit code 0

进程池的使用

apply(self, func, args=(), kwds={}) # 使用arg和kwds参数调用func函数，结果返回前会一直阻塞，这样就导致子进程会顺序执行，而不是并发执行

apply_async(self, func, args=(), kwds={}, callback=None,error_callback=None)#  apply()方法的一个变体，会返回一个结果对象。如果callback被指定，那么callback可以接收一个参数然后被调用，当结果准备好回调时会调用callback，调用失败时，则用error_callback替换callback。 Callbacks应被立即完成，否则处理结果的线程会被阻塞。对比apply方法，该方法不会阻塞，类似线程的setDaemon

pool.close() # 阻止更多的任务提交到pool，待任务完成后，工作进程会退出

pool.terminate() # 不管任务是否完成，立即停止工作进程。在对pool对象进程垃圾回收的时候，会立即调用terminate()。

pool.join()  # wait工作线程的退出，在调用join()前，必须调用close() or terminate()。这样是因为被终止的进程需要被父进程调用wait（join等价与wait），否则进程会成为僵尸进程。

参考:https://docs.python.org/3/library/multiprocessing.html#module-multiprocessing.pool

协程

概念

　　线程和进程的操作是由程序触发系统接口，最后的执行者是系统；协程的操作则是程序员。

　　协程存在的意义：对于多线程应用，CPU通过切片的方式来切换线程间的执行，线程切换时需要耗时（保存状态，下次继续）。协程，则只使用一个线程，在一个线程中规定某个代码块执行顺序。

　　协程的适用场景：当程序中存在大量不需要CPU的操作时（IO），适用于协程；

实现

　　协程的实现主要借助2个模块，greenlet和gevent(内部调用greenlet)

import gevent
def fun1():
  print("www.baidu.com")  # 第一步
  gevent.sleep(0)
  print("end the baidu.com") # 第三步
def fun2():
  print("www.zhihu.com")  # 第二步
  gevent.sleep(0)
  print("end th zhihu.com") # 第四步
gevent.joinall([
  gevent.spawn(fun1),
  gevent.spawn(fun2),
])

gevent

import greenlet
def fun1():
  print("") # 第一步
  gr2.switch()
  print("")  # 第三步
  gr2.switch()
def fun2():
  print("") # 第二步
  gr1.switch()
  print("") # 第四步
gr1 = greenlet.greenlet(fun1)
gr2 = greenlet.greenlet(fun2)
gr1.switch()

greenlet

from gevent import monkey; monkey.patch_all()
import gevent
import urllib2

def f(url):
    print('GET: %s' % url)
    resp = urllib2.urlopen(url)
    data = resp.read()
    print('%d bytes received from %s.' % (len(data), url))

gevent.joinall([
        gevent.spawn(f, 'https://www.python.org/'),
        gevent.spawn(f, 'https://www.yahoo.com/'),
        gevent.spawn(f, 'https://github.com/'),
])

遇到IO操作自动切换

event loop是协程执行的控制点， 如果你希望执行协程， 就需要用到它们。
event loop提供了如下的特性：
•注册、执行、取消延时调用(异步函数)
•创建用于通信的client和server协议(工具)
•创建和别的程序通信的子进程和协议(工具)
•把函数调用送入线程池中

import asyncio
async def cor1():
  print("COR1 start")
  await cor2()
  print("COR1 end")
async def cor2():
  print("COR2")
loop = asyncio.get_event_loop()
loop.run_until_complete(cor1())
loop.close()

out:
COR1 start
COR2
COR1 end

asyncio

最后三行是重点。
•asyncio.get_event_loop() : asyncio启动默认的event loop
•run_until_complete() : 这个函数是阻塞执行的，知道所有的异步函数执行完成，
•close() : 关闭event loop。

额外扩展：Timer

　　定时器，指定n秒后执行某操作

from threading import Timer

def hello():
    print("hello, world")

t = Timer(1, hello)
t.start()  # after 1 seconds, "hello, world" will be printed