import threading

import time

def countNum(n): # 定义某个线程要运行的函数

    print("running on number:%s" %n)

    time.sleep(3)

if __name__ == '__main__':

    t1 = threading.Thread(target=countNum,args=(23,)) #生成一个线程实例

    t2 = threading.Thread(target=countNum,args=(34,))

    t1.start() #启动线程

    t2.start()

    print("ending!")

#继承Thread式创建

import threading

import time

class MyThread(threading.Thread):

    def __init__(self,num):

        threading.Thread.__init__(self)

        self.num=num

    def run(self):

        print("running on number:%s" %self.num)

        time.sleep(3)

t1=MyThread(56)

t2=MyThread(78)

t1.start()

t2.start()

print("ending")

# join()：在子线程完成运行之前，这个子线程的父线程将一直被阻塞。

# setDaemon(True)：

        '''

         将线程声明为守护线程，必须在start() 方法调用之前设置，如果不设置为守护线程程序会被无限挂起。

         当我们在程序运行中，执行一个主线程，如果主线程又创建一个子线程，主线程和子线程 就分兵两路，分别运行，那么当主线程完成

         想退出时，会检验子线程是否完成。如果子线程未完成，则主线程会等待子线程完成后再退出。但是有时候我们需要的是只要主线程

         完成了，不管子线程是否完成，都要和主线程一起退出，这时就可以 用setDaemon方法啦'''

import threading

from time import ctime,sleep

import time

def Music(name):

        print ("Begin listening to {name}. {time}".format(name=name,time=ctime()))

        sleep(3)

        print("end listening {time}".format(time=ctime()))

def Blog(title):

        print ("Begin recording the {title}. {time}".format(title=title,time=ctime()))

        sleep(5)

        print('end recording {time}'.format(time=ctime()))

threads = []

t1 = threading.Thread(target=Music,args=('FILL ME',))

t2 = threading.Thread(target=Blog,args=('',))

threads.append(t1)

threads.append(t2)

if __name__ == '__main__':

    #t2.setDaemon(True)

    for t in threads:

        #t.setDaemon(True) #注意:一定在start之前设置

        t.start()

        #t.join()

    #t1.join()

    #t2.join()    #  考虑这三种join位置下的结果？

    print ("all over %s" %ctime())

daemon

A boolean value indicating whether this thread is a daemon thread (True) or not (False). This must be set before start() is called, otherwise RuntimeError is raised. Its initial value is inherited from the creating thread; the main thread is not a daemon thread and therefore all threads created in the main thread default to daemon = False.

The entire Python program exits when no alive non-daemon threads are left.

当daemon被设置为True时，如果主线程退出，那么子线程也将跟着退出，

反之，子线程将继续运行，直到正常退出。

'''

定义：

In CPython, the global interpreter lock, or GIL, is a mutex that prevents multiple

native threads from executing Python bytecodes at once. This lock is necessary mainly

because CPython’s memory management is not thread-safe. (However, since the GIL

exists, other features have grown to depend on the guarantees that it enforces.)

'''

'''

2.3.1 GIL的早期设计

Python支持多线程，而解决多线程之间数据完整性和状态同步的最简单方法自然就是加锁。 于是有了GIL这把超级大锁，而当越来越多的代码库开发者接受了这种设定后，他们开始大量依赖这种特性（即默认python内部对象是thread-safe的，无需在实现时考虑额外的内存锁和同步操作）。慢慢的这种实现方式被发现是蛋疼且低效的。但当大家试图去拆分和去除GIL的时候，发现大量库代码开发者已经重度依赖GIL而非常难以去除了。有多难？做个类比，像MySQL这样的“小项目”为了把Buffer Pool Mutex这把大锁拆分成各个小锁也花了从5.5到5.6再到5.7多个大版为期近5年的时间，并且仍在继续。MySQL这个背后有公司支持且有固定开发团队的产品走的如此艰难，那又更何况Python这样核心开发和代码贡献者高度社区化的团队呢？

2.3.2 GIL的影响

无论你启多少个线程，你有多少个cpu, Python在执行一个进程的时候会淡定的在同一时刻只允许一个线程运行。

所以，python是无法利用多核CPU实现多线程的。

这样，python对于计算密集型的任务开多线程的效率甚至不如串行(没有大量切换)，但是，对于IO密集型的任务效率还是有显著提升的。

'''

用multiprocessing替代Thread multiprocessing库的出现很大程度上是为了弥补thread库因为GIL而低效的缺陷。它完整的复制了一套thread所提供的接口方便迁移。唯一的不同就是它使用了多进程而不是多线程。每个进程有自己的独立的GIL，因此也不会出现进程之间的GIL争抢。

#coding:utf8

from multiprocessing import Process

import time

def counter():

    i = 0

    for _ in range(40000000):

        i = i + 1

    return True

def main():

    l=[]

    start_time = time.time()

    for _ in range(2):

        t=Process(target=counter)

        t.start()

        l.append(t)

        #t.join()

    for t in l:

       t.join()

    end_time = time.time()

    print("Total time: {}".format(end_time - start_time))

if __name__ == '__main__':

    main()

'''

py2.7:

     串行:6.1565990448 s

     并行:3.1639978885 s

py3.5:

     串行:6.556925058364868 s

     并发:3.5378448963165283 s

'''

当然multiprocessing也不是万能良药。它的引入会增加程序实现时线程间数据通讯和同步的困难。就拿计数器来举例子，如果我们要多个线程累加同一个变量，对于thread来说，申明一个global变量，用thread.Lock的context包裹住三行就搞定了。而multiprocessing由于进程之间无法看到对方的数据，只能通过在主线程申明一个Queue，put再get或者用share memory的方法。这个额外的实现成本使得本来就非常痛苦的多线程程序编码，变得更加痛苦了。

总结：因为GIL的存在，只有IO Bound场景下得多线程会得到较好的性能 - 如果对并行计算性能较高的程序可以考虑把核心部分也成C模块，或者索性用其他语言实现 - GIL在较长一段时间内将会继续存在，但是会不断对其进行改进。

所谓死锁： 是指两个或两个以上的进程或线程在执行过程中，因争夺资源而造成的一种互相等待的现象，若无外力作用，它们都将无法推进下去。此时称系统处于死锁状态或系统产生了死锁，这些永远在互相等待的进程称为死锁进程。

import threading

import time

mutexA = threading.Lock()

mutexB = threading.Lock()

class MyThread(threading.Thread):

    def __init__(self):

        threading.Thread.__init__(self)

    def run(self):

        self.fun1()

        self.fun2()

    def fun1(self):

        mutexA.acquire()  # 如果锁被占用,则阻塞在这里,等待锁的释放

        print ("I am %s , get res: %s---%s" %(self.name, "ResA",time.time()))

        mutexB.acquire()

        print ("I am %s , get res: %s---%s" %(self.name, "ResB",time.time()))

        mutexB.release()

        mutexA.release()

    def fun2(self):

        mutexB.acquire()

        print ("I am %s , get res: %s---%s" %(self.name, "ResB",time.time()))

        time.sleep(0.2)

        mutexA.acquire()

        print ("I am %s , get res: %s---%s" %(self.name, "ResA",time.time()))

        mutexA.release()

        mutexB.release()

if __name__ == "__main__":

    print("start---------------------------%s"%time.time())

    for i in range(0, 10):

        my_thread = MyThread()

        my_thread.start()

'''

结果就是，线程一获取ab锁之后先后释放，然后线程二获得a锁，线程一获得b锁，然后线程一需要a锁，二需要b锁，然后就两个进程僵持住了

'''

在Python中为了支持在同一线程中多次请求同一资源，python提供了可重入锁RLock。这个RLock内部维护着一个Lock和一个counter变量，counter记录了acquire的次数，从而使得资源可以被多次require。直到一个线程所有的acquire都被release，其他的线程才能获得资源。上面的例子如果使用RLock代替Lock，则不会发生死锁：（类似递归）

2.8 队列(queue)

queue is especially useful in threaded programming when information must be exchanged safely between multiple threads.

2.8.1 get与put方法

'''

创建一个“队列”对象

import queue

q = queue.Queue(maxsize = 10)

queue.Queue类即是一个队列的同步实现。队列长度可为无限或者有限。可通过Queue的构造函数的可选参数

maxsize来设定队列长度。如果maxsize小于1就表示队列长度无限。

将一个值放入队列中

q.put(10)

调用队列对象的put()方法在队尾插入一个项目。put()有两个参数，第一个item为必需的，为插入项目的值；

第二个block为可选参数，默认为

1。如果队列当前为空且block为1，put()方法就使调用线程暂停,直到空出一个数据单元。如果block为0，

put方法将引发Full异常。

将一个值从队列中取出

q.get()

调用队列对象的get()方法从队头删除并返回一个项目。可选参数为block，默认为True。如果队列为空且

block为True，get()就使调用线程暂停，直至有项目可用。如果队列为空且block为False，队列将引发Empty异常。

'''

2.8.2  join与task_done方法

'''

put 和 task_done 一一对应，表示所有曾加到队列中的任务，都执行完了，然后才能执行join之后的操作

'''

'''

join() 阻塞进程，直到所有任务完成，需要配合另一个方法task_done。

    def join(self):

     with self.all_tasks_done:

      while self.unfinished_tasks:

       self.all_tasks_done.wait()

task_done() 表示某个任务完成。每一条get语句后需要一条task_done。

import queue

q = queue.Queue(5)

q.put(10)

q.put(20)

print(q.get())

q.task_done()

print(q.get())

q.task_done()

q.join()

print("ending!")

'''

2.8.3 其他常用方法

'''

此包中的常用方法(q = Queue.Queue()):

q.qsize() 返回队列的大小

q.empty() 如果队列为空，返回True,反之False

q.full() 如果队列满了，返回True,反之False

q.full 与 maxsize 大小对应

q.get([block[, timeout]]) 获取队列，timeout等待时间

q.get_nowait() 相当q.get(False)非阻塞 
q.put(item) 写入队列，timeout等待时间

q.put_nowait(item) 相当q.put(item, False)

q.task_done() 在完成一项工作之后，q.task_done() 函数向任务已经完成的队列发送一个信号

q.join() 实际上意味着等到队列为空，再执行别的操作

'''

2.8.4 其他模式

'''

Python Queue模块有三种队列及构造函数: 

1、Python Queue模块的FIFO队列先进先出。  class queue.Queue(maxsize)

2、LIFO类似于堆，即先进后出。           class queue.LifoQueue(maxsize)

3、还有一种是优先级队列级别越低越先出来。 class queue.PriorityQueue(maxsize)

'''

import queue

#先进后出

q=queue.LifoQueue()

q.put(34)

q.put(56)

q.put(12)

#优先级

q=queue.PriorityQueue()

q.put([5,100])

q.put([7,200])

q.put([3,"hello"])

q.put([4,{"name":"alex"}])

while 1:

  data=q.get()

  print(data)

2.8.5 生产者消费者模型

在线程世界里，生产者就是生产数据的线程，消费者就是消费数据的线程。在多线程开发当中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。

生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。

这就像，在餐厅，厨师做好菜，不需要直接和客户交流，而是交给前台，而客户去饭菜也不需要不找厨师，直接去前台领取即可，这也是一个结耦的过程。

import time,random

import queue,threading

q = queue.Queue()

def Producer(name):

  count = 0

  while count <10:

    print("making........")

    time.sleep(random.randrange(3))

    q.put(count)

    print('Producer %s has produced %s baozi..' %(name, count))

    count +=1

    #q.task_done()

    #q.join()

    print("ok......")

def Consumer(name):

  count = 0

  while count <10:

    time.sleep(random.randrange(4))

    if not q.empty():

        data = q.get()

        #q.task_done()

        #q.join()

        print(data)

        print('\033[32;1mConsumer %s has eat %s baozi...\033[0m' %(name, data))

    else:

        print("-----no baozi anymore----")

    count +=1

p1 = threading.Thread(target=Producer, args=('A',))

c1 = threading.Thread(target=Consumer, args=('B',))

# c2 = threading.Thread(target=Consumer, args=('C',))

# c3 = threading.Thread(target=Consumer, args=('D',))

p1.start()

c1.start()

# c2.start()

# c3.start()

三 multiprocessing模块

'''

Multiprocessing is a package that supports spawning processes using an API similar to the threading module. The multiprocessing package offers both local and remote concurrency,effectively side-stepping the Global Interpreter Lock by using subprocesses instead of threads. Due to this, the multiprocessing module allows the programmer to fully leverage multiple processors on a given machine. It runs on both Unix and Windows.

由于GIL的存在，python中的多线程其实并不是真正的多线程，如果想要充分地使用多核CPU的资源，在python中大部分情况需要使用多进程。

multiprocessing包是Python中的多进程管理包。与threading.Thread类似，它可以利用multiprocessing.Process对象来创建一个进程。该进程可以运行在Python程序内部编写的函数。该Process对象与Thread对象的用法相同，也有start(), run(), join()的方法。此外multiprocessing包中也有Lock/Event/Semaphore/Condition类 (这些对象可以像多线程那样，通过参数传递给各个进程)，用以同步进程，其用法与threading包中的同名类一致。所以，multiprocessing的很大一部份与threading使用同一套API，只不过换到了多进程的情境。

'''

# Process类调用

from multiprocessing import Process

import time

def f(name):

    print('hello', name,time.ctime())

    time.sleep(1)

if __name__ == '__main__':

    p_list=[]

    for i in range(3):

        p = Process(target=f, args=('alvin:%s'%i,))

        p_list.append(p)

        p.start()

    for i in p_list:

        p.join()

    print('end')

# 继承Process类调用

from multiprocessing import Process

import time

class MyProcess(Process):

    def __init__(self):

        super(MyProcess, self).__init__()

        # self.name = name

    def run(self):

        print ('hello', self.name,time.ctime())

        time.sleep(1)

if __name__ == '__main__':

    p_list=[]

    for i in range(3):

        p = MyProcess()

        p.start()

        p_list.append(p)

    for p in p_list:

        p.join()

    print('end')

3.2 process类

需要写在 if __name__ == '__main__': 内

构造方法：

Process([group [, target [, name [, args [, kwargs]]]]])

　　group: 线程组，目前还没有实现，库引用中提示必须是None； 
　　target: 要执行的方法； 
　　name: 进程名； 
　　args/kwargs: 要传入方法的参数。

实例方法：

　　is_alive()：返回进程是否在运行。

　　join([timeout])：阻塞当前上下文环境的进程程，直到调用此方法的进程终止或到达指定的timeout（可选参数）。

　　start()：进程准备就绪，等待CPU调度

　　run()：strat()调用run方法，如果实例进程时未制定传入target，这star执行t默认run()方法。

　　terminate()：不管任务是否完成，立即停止工作进程

属性：

　　daemon：和线程的setDeamon功能一样

　　name：进程名字。

　　pid：进程号。

from multiprocessing import Process

import os

import time

def info(name):

    print("name:",name)

    print('parent process:', os.getppid())

    print('process id:', os.getpid())

    print("------------------")

    time.sleep(1)

def foo(name):

    info(name)

if __name__ == '__main__':

    info('main process line')

    p1 = Process(target=info, args=('alvin',))

    p2 = Process(target=foo, args=('egon',))

    p1.start()

    p2.start()

    p1.join()

    p2.join()

    print("ending")

3.3 进程间通讯　

3.3.1 进程对列Queue

from multiprocessing import Process, Queue

import queue

def f(q,n):

    #q.put([123, 456, 'hello'])

    q.put(n*n+1)

    print("son process",id(q))

if __name__ == '__main__':

    q = Queue()  #try: q=queue.Queue()  #在main内定义

    print("main process",id(q))

    for i in range(3):

        p = Process(target=f, args=(q,i))

        p.start()

    print(q.get())

    print(q.get())

    print(q.get())

3.3.2 管道(pipe)

The Pipe() function returns a pair of connection objects connected by a pipe which by default is duplex (two-way). For example:

from multiprocessing import Process, Pipe

def f(conn):

    conn.send([12, {"name":"yuan"}, 'hello'])

    response=conn.recv()

    print("response",response)

    conn.close()

    if __name__ == '__main__':

        parent_conn, child_conn = Pipe()

        p = Process(target=f, args=(child_conn,))

        p.start()

        print(parent_conn.recv())   # prints "[42, None, 'hello']"

        parent_conn.send("儿子你好!")

        p.join()

Pipe（）返回的两个连接对象代表管道的两端。 每个连接对象都有send()和recv()方法（等等）。 请注意，如果两个进程（或线程）尝试同时读取或写入管道的同一端，管道中的数据可能会损坏。

3.3.3 manager（Manager支持的类型有list,dict,Namespace,Lock,RLock,Semaphore,BoundedSemaphore,Condition,Event,Queue,Value和Array。 ）

Queue和pipe只是实现了数据交互，并没实现数据共享，即一个进程去更改另一个进程的数据。

A manager object returned by Manager() controls a server process which holds Python objects and allows other processes to manipulate them using proxies.

from multiprocessing import Process, Manager

def f(d, l,n):

    d[n] = n

    d["name"] ="alvin"

    l.append(n)

    #print("l",l)

if __name__ == '__main__':

    with Manager() as manager:

        d = manager.dict()

        l = manager.list(range(5))

        p_list = []

        for i in range(10):

            p = Process(target=f, args=(d,l,i))

            p.start()

            p_list.append(p)

        for res in p_list:

            res.join()

        print(d)

        print(l)

3.4 进程池

进程池内部维护一个进程序列，当使用时，则去进程池中获取一个进程，如果进程池序列中没有可供使用的进进程，那么程序就会等待，直到进程池中有可用进程为止。

进程池内部维护一个进程序列，当使用时，去进程池中获取一个进程，如果进程池序列中没有可供使用的进程，那么程序就会等待，直到进程池中有可用进程为止。

from multiprocessing import Pool

import time

def foo(args):

 time.sleep(1)

 print(args)

if __name__ == '__main__':

 p = Pool(5)

 for i in range(30):

     p.apply_async(func=foo, args= (i,))

 p.close()   # 等子进程执行完毕后关闭线程池

 # time.sleep(2)

 # p.terminate()  # 立刻关闭线程池

 p.join()

进程池中有以下几个主要方法：

apply：从进程池里取一个进程并执行

apply_async：apply的异步版本

terminate:立刻关闭线程池

join：主进程等待所有子进程执行完毕，必须在close或terminate之后

close：等待所有进程结束后，才关闭线程池