python 线程、进程、GIL锁、多线程及异步交互、多进程及进程间通信_python多线程gil异步执行

 

在上文中我们学习了线程与进程，那么python中是如何处理的？

先看下单进程执行的情况

import time
def run(n):
    print("task ",n )
    time.sleep(2)
 
run("t1")
run("t2")

在上面程序中，简单调用两次run方法，该方法会延时2s，输出结果：输出task t1后隔2s，输出task t2，2s后程序结束

如何让这两个方法同时执行呢，这就需要给每个方法启用一个单独的线程

1、启动多个线程

import threading
import time
def run(n):
    print("task ",n )
    time.sleep(2)
 
# run("t1")
# run("t2")
t1 = threading.Thread(target=run,args=("t1",))#生成一个线程实例
t2 = threading.Thread(target=run,args=("t2",))
t1.start()
t2.start()

由于每个方法都有单独的进程，会同时开始执行，上面代码会同时输出

task  t1
task  t2

等待2s后程序结束

除了这种简单实现，还可以自己定义类，继承调用

import threading
import time
 
class MyThread(threading.Thread):
    def __init__(self,n,sleep_time):
        super(MyThread,self).__init__()
        self.n =  n
        self.sleep_time = sleep_time
    def run(self):
        print("runnint task ",self.n )
        time.sleep(self.sleep_time)
        print("task done,",self.n )
 
t1 = MyThread("t1",2)
t2 = MyThread("t2",4)
t1.start()
t2.start()

如何启动多个进程

import threading
import time
def run(n):
    print("task ",n )
    time.sleep(2)
    print("task done",n)
 
start_time = time.time()
for i in range(50):
    t = threading.Thread(target=run,args=("t-%s" %i ,))
    t.start()
 
print("----------all threads has finished...")
print("cost:",time.time() - start_time)

上面的代码在同时启动50个线程后，会直接输出

----------all threads has finished...

cost：0.0080

发现主线程直接结束，没有等子线程，2s后子线程分别task done

代码共有51个线程，一个主线程与50个子线程，主线程无法计算子线程执行时间

那么，我们如何计算所有线程执行时间？

此时需要设置主线程等待子线程执行结束，通过一个临时列表，在线程启动后分别join等待，子线程分别结束后，结束主进程，计算耗时约2.011415958s

import threading
import time
def run(n):
    print("task ",n )
    time.sleep(2)
    print("task done",n)
 
start_time = time.time()
t_objs = [] #存线程实例
for i in range(50):
    t = threading.Thread(target=run,args=("t-%s" %i ,))
    t.start()
    t_objs.append(t) #为了不阻塞后面线程的启动，不在这里join，先放到一个列表里
 
for t in t_objs: #循环线程实例列表，等待所有线程执行完毕
    t.join()
print("----------all threads has finished...")
print("cost:",time.time() - start_time)

2、守护线程Daemon

Some threads do background tasks, like sending keepalive packets, or performing periodic garbage collection, or whatever. These are only useful when the main program is running, and it's okay to kill them off once the other, non-daemon, threads have exited.

Without daemon threads, you'd have to keep track of them, and tell them to exit, before your program can completely quit. By setting them as daemon threads, you can let them run and forget about them, and when your program quits, any daemon threads are killed automatically.

join会让主线程等待子线程结束后再结束，将子线程变成守护线程后，主进程就不需要等待了

这个设置要在启动线程start之前，启动线程后就不可以设置为守护线程了

程序会等主线程结束，但不会等守护线程

import threading
import time
def run(n):
    print("task ",n )
    time.sleep(2)
    print("task done",n,threading.current_thread())
 
start_time = time.time()
t_objs = [] #存线程实例
for i in range(50):
    t = threading.Thread(target=run,args=("t-%s" %i ,))
    t.setDaemon(True) #把当前线程设置为守护线程
    t.start()
    t_objs.append(t) #为了不阻塞后面线程的启动，不在这里join，先放到一个列表里
 
# for t in t_objs: #循环线程实例列表，等待所有线程执行完毕
#     t.join()
 
# time.sleep(2)
print("----------all threads has finished...",threading.current_thread(),threading.active_count())
print("cost:",time.time() - start_time)

输出：----------all threads has finished... <_MainThread(MainThread, started 140736069915584)> 51
cost: 0.0067479610443115234

3、Python GIL(Global Interpreter Lock)　　

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.)

我们现在的机器一般是多核的，如四核八核等，若单核机器，那么一定是串行执行，多核则可以同时执行不同的任务

但是在python中，同一时间执行的线程只有一个，看起来是并发是由于在进行上下文切换

首先需要明确的一点是GIL(全局解释器锁)并不是Python的特性，它是在实现Python解析器(CPython)时所引入的一个概念。就好比C++是一套语言（语法）标准，但是可以用不同的编译器来编译成可执行代码。有名的编译器例如GCC，INTEL C++，Visual C++等。Python也一样，同样一段代码可以通过CPython，PyPy，Psyco等不同的Python执行环境来执行。像其中的JPython就没有GIL。然而因为CPython是大部分环境下默认的Python执行环境。所以在很多人的概念里CPython就是Python，也就想当然的把GIL归结为Python语言的缺陷。所以这里要先明确一点：GIL并不是Python的特性，Python完全可以不依赖于GIL

这篇文章透彻的剖析了GIL对python多线程的影响，强烈推荐看一下：http://www.dabeaz.com/python/UnderstandingGIL.pdf 

import time
import threading
def addNum():
    global num #在每个线程中都获取这个全局变量
    print('--get num:',num )
    time.sleep(1)
    num  -=1 #对此公共变量进行-1操作
num = 100  #设定一个共享变量
thread_list = []
for i in range(100):
    t = threading.Thread(target=addNum)
    t.start()
    thread_list.append(t)
for t in thread_list: #等待所有线程执行完毕
    t.join()
 
print('final num:', num )

先看上面的代码，正常来讲，这个num结果应该是0， 但在python 2.7上多运行几次，会发现，最后打印出来的num结果不总是0，为什么每次运行的结果不一样呢？ 假设你有A,B两个线程，此时都 要对num 进行减1操作， 由于2个线程是并发同时运行的，所以2个线程很有可能同时拿走了num=100这个初始变量交给cpu去运算，当A线程去处完的结果是99，但此时B线程运算完的结果也是99，两个线程同时CPU运算的结果再赋值给num变量后，结果就都是99。

python解释器CPython是用C实现的，在有GIL的情况下，仍然会发生上面的情况，为了解决，可以再加锁，保证同一时间只有一个线程修改数据

每个线程在要修改公共数据时，为了避免自己在还没改完的时候别人也来修改此数据，可以给这个数据加一把锁， 这样其它线程想修改此数据时就必须等待你修改完毕并把锁释放掉后才能再访问此数据。 

*注：3.x上的结果总是正确的，可能是自动加了锁。以上这个问题在CPython上存在，PyPy和Jpython不存在这个问题

线程同步

如果多个线程共同对某个数据修改，则可能出现不可预料的结果，为了保证数据的正确性，需要对多个线程进行同步。

使用 Thread 对象的 Lock 和 Rlock 可以实现简单的线程同步，这两个对象都有 acquire 方法和 release 方法，对于那些需要每次只允许一个线程操作的数据，可以将其操作放到 acquire 和 release 方法之间。

加全局锁的情况：互斥锁

import time
import threading
def addNum():
    global num  # 在每个线程中都获取这个全局变量
    print('--get num:', num)
    lock.acquire()  # 修改数据前加锁
    # time.sleep(1)
    num -= 1  # 对此公共变量进行-1操作
    lock.release()  # 修改后释放
 
num = 100  # 设定一个共享变量
thread_list = []
lock = threading.Lock()  # 生成全局锁
for i in range(100):
    t = threading.Thread(target=addNum)
    t.start()
    thread_list.append(t)
for t in thread_list:  # 等待所有线程执行完毕
    t.join()
 
print('final num:', num)

GIL VS Lock 

Python既然有一个GIL来保证同一时间只能有一个线程来执行了，为什么这里还需要lock? 注意啦，这里的lock是用户级的lock,跟那个GIL没关系 ，具体我们通过下图来看一下，就明白了。
 

那你又问了， 既然用户程序已经自己有锁了，那为什么C python还需要GIL呢？加入GIL主要的原因是为了降低程序的开发的复杂度，比如现在的你写python不需要关心内存回收的问题，因为Python解释器帮你自动定期进行内存回收，你可以理解为python解释器里有一个独立的线程，每过一段时间它起wake up做一次全局轮询看看哪些内存数据是可以被清空的，此时你自己的程序 里的线程和 py解释器自己的线程是并发运行的，假设你的线程删除了一个变量，py解释器的垃圾回收线程在清空这个变量的过程中的clearing时刻，可能一个其它线程正好又重新给这个还没来及得清空的内存空间赋值了，结果就有可能新赋值的数据被删除了，为了解决类似的问题，python解释器简单粗暴的加了锁，即当一个线程运行时，其它人都不能动，这样就解决了上述的问题，  这可以说是Python早期版本的遗留问题。

4，递归锁RLock

当锁内还需要加锁的时候，需要用到递归锁，否则会出现锁死的情况，大锁内套小锁的问题

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

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()

#!/usr/bin/env python
# _*_ coding:utf-8 _*_
import threading, time
def run1():
    print("grab the first part data")
    lock.acquire()
    global num
    num += 1
    lock.release()
    return num
 
def run2():
    print("grab the second part data")
    lock.acquire()
    global num2
    num2 += 1
    lock.release()
    return num2
 
def run3():
    lock.acquire()
    res = run1()
    print('--------between run1 and run2-----')
    res2 = run2()
    lock.release()
    print(res, res2)
 
if __name__ == '__main__':
    num, num2 = 0, 0
    lock = threading.RLock()
    for i in range(10):
        t = threading.Thread(target=run3)
        t.start()
 
while threading.active_count() != 1:
    print(threading.active_count())
else:
    print('----all threads done---')
    print(num, num2)

5、Semaphore(信号量)

互斥锁 同时只允许一个线程更改数据，而Semaphore是同时允许一定数量的线程更改数据 ，比如房间有3个座位，那最多只允许3个人进入房间，后面的人只能等里面有人出来了才能再进去。

#!/usr/bin/env python
# _*_ coding:utf-8 _*_
import threading, time
def run(n):
    semaphore.acquire()
    time.sleep(1)
    print("run the thread: %s\n" % n)
    semaphore.release()
 
if __name__ == '__main__':
    semaphore = threading.BoundedSemaphore(5)  # 最多允许5个线程同时运行
    for i in range(20):
        t = threading.Thread(target=run, args=(i,))
        t.start()
 
while threading.active_count() != 1:
    pass  # print threading.active_count()
else:
    print('----all threads done---')

条件（Condition）

可以把Condition理解为一把高级的琐，它提供了比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() 
唤醒所有挂起的线程（如果存在挂起的线程）。注意：这些方法不会释放所占用的琐。

使得线程等待，只有满足某条件时，才释放n个线程

import threading
  
def run(n):
    con.acquire()
    con.wait()
    print("run the thread: %s" %n)
    con.release()
  
if __name__ == '__main__':
  
    con = threading.Condition()
    for i in range(10):
        t = threading.Thread(target=run, args=(i,))
        t.start()
  
    while True:
        inp = input('>>>')
        if inp == 'q':
            break
        con.acquire()
        con.notify(int(inp))
        con.release()<br><br>def condition_func():<br><br>    ret = False<br>    inp = input('>>>')<br>    if inp == '1':<br>        ret = True<br><br>    return ret<br><br><br>def run(n):<br>    con.acquire()<br>    con.wait_for(condition_func)<br>    print("run the thread: %s" %n)<br>    con.release()<br><br>if __name__ == '__main__':<br><br>    con = threading.Condition()<br>    for i in range(10):<br>        t = threading.Thread(target=run, args=(i,))<br>        t.start()

Timer

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

def hello():
    print("hello, world")
t = Timer(30.0, hello)
t.start()  # after 30 seconds, "hello, world" will be printed

6、事件Events

An event is a simple synchronization同步 object;

the event represents an internal flag, and threads can wait for the flag to be set, or set or clear the flag themselves.
event = threading.Event()
# a client thread can wait for the flag to be set
event.wait()
# a server thread can set or reset it
event.set()
event.clear()
If the flag is set, the wait method doesn’t do anything.
If the flag is cleared, wait will block until it becomes set again.
Any number of threads may wait for the same event.

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

事件处理的机制：全局定义了一个“Flag”，如果“Flag”值为 False，那么当程序执行 event.wait 方法时就会阻塞，如果“Flag”值为True，那么event.wait 方法时便不再阻塞。

• clear：将“Flag”设置为False
• set：将“Flag”设置为True
• Event.isSet() ：判断标识位是否为Ture。

通过Event来实现两个或多个线程间的交互，下面是一个红绿灯的例子，即起动一个线程做交通指挥灯，生成几个线程做车辆，车辆行驶按红灯停，绿灯行的规则。

 这里还有一个event使用的例子，员工进公司门要刷卡， 我们这里设置一个线程是“门”， 再设置几个线程为“员工”，员工看到门没打开，就刷卡，刷完卡，门开了，员工就可以通过。

#!/usr/bin/env python
# _*_ coding:utf-8 _*_
import threading,time
import random
def light():
    if not event.isSet():
        event.set() #wait就不阻塞 #绿灯状态
    count = 0
    while True:
        if count < 10:
            print('\033[42;1m--green light on---\033[0m')
        elif count <13:
            print('\033[43;1m--yellow light on---\033[0m')
        elif count <20:
            if event.isSet():
                event.clear()
            print('\033[41;1m--red light on---\033[0m')
        else:
            count = 0
            event.set() #打开绿灯
        time.sleep(1)
        count +=1
def car(n):
    while 1:
        time.sleep(random.randrange(10))
        if  event.isSet(): #绿灯
            print("car [%s] is running.." % n)
        else:
            print("car [%s] is waiting for the red light.." %n)
if __name__ == '__main__':
    event = threading.Event()
    Light = threading.Thread(target=light)
    Light.start()
    for i in range(3):
        t = threading.Thread(target=car,args=(i,))
        t.start()

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import threading
import time
import random
def door():
    door_open_time_counter = 0
    # if not door_swiping_event.isSet():
    #     door_swiping_event.set()   #刚开始设置为开门状态
    # if door_swiping_event.isSet():
    #     door_swiping_event.clear() #刚开始设置为关门状态，默认也是开始时候关门状态
    while True:
        if door_swiping_event.is_set():
            print("\033[32;1mdoor opening....\033[0m")
            door_open_time_counter +=1
        else:
            print("\033[31;1mdoor closed...., swipe to open.\033[0m")
            door_open_time_counter = 0 #清空计时器
            door_swiping_event.wait()
        if door_open_time_counter > 3: #门开了已经3s了,该关了
            door_swiping_event.clear()
        time.sleep(0.5)
 
def staff(n):
    print("staff [%s] is comming..." % n )
    while True:
        if door_swiping_event.is_set():
            print("\033[34;1mdoor is opened, passing.....\033[0m")
            break
        else:
            print("staff [%s] sees door got closed, swipping the card....." % n)
            # print(door_swiping_event.set())
            door_swiping_event.set()
            print("after set ",door_swiping_event.set())
        time.sleep(0.5)
 
door_swiping_event  = threading.Event() #设置事件
door_thread = threading.Thread(target=door)
door_thread.start()
for i in range(5):
    p = threading.Thread(target=staff,args=(i,))
    time.sleep(random.randrange(3))
    p.start()

 

7、Queue消息队列 实现多线程间异步交互

Queue是python标准库中的线程安全的队列（FIFO）实现,提供了一个适用于多线程编程的先进先出的数据结构，即队列，用来在生产者和消费者线程之间的信息传递，在多线程间安全地交换数据。

作用：1，解耦：使程序直接实现松耦合，修改一个函数，不会有串联关系。2，提高处理效率。

class queue.Queue(maxsize=0) #先入先出FIFO maxsize 可设置大小，设置block=False抛异常

class queue.LifoQueue(maxsize=0) #last in fisrt out 

class queue.PriorityQueue(maxsize=0) #存储数据时可设置优先级的队列优先级设置数越小等级越高

Constructor for a priority queue. maxsize is an integer that sets the upperbound limit on the number of items that can be placed in the queue. Insertion will block once this size has been reached, until queue items are consumed. If maxsize is less than or equal to zero, the queue size is infinite.

The lowest valued entries are retrieved first (the lowest valued entry is the one returned by sorted(list(entries))[0]). A typical pattern for entries is a tuple in the form: (priority_number, data).

队列的简单应用，队列定义

import Queue
q = Queue.Queue()
for i in range(5):
    q.put(i)
while not q.empty():
#while q.qsize > 0:
    print q.get()

若q中存的已经被get取完，继续get取，程序会卡住，get_nowait()在取完继续取时候会抛出异常

为了防止队列中取完再取的情况，用q.enmpty判断还可以用q,qsize是否为0来判断

队列的方法：

# 放入数据
Queue.put(item, block=True, timeout=None)
将item放入队列中。如果可选的参数block为真且timeout为空对象（默认的情况，阻塞调用，无超时），如有必要（比如队列满），阻塞调用线程，直到有空闲槽可用。如果timeout是个正整数，阻塞调用进程最多timeout秒，如果一直无空闲槽可用，抛出Full异常（带超时的阻塞调用）。如果block为假，如果有空闲槽可用将数据放入队列，否则立即抛出Full异常（非阻塞调用，timeout被忽略）

# 取出数据 #没有数据将会等待
Queue.get(block=True, timeout=None)
从队列中移除并返回一个数据。如果可选的参数block为真且timeout为空对象（默认的情况，阻塞调用，无超时），阻塞调用进程直到有数据可用。如果timeout是个正整数，阻塞调用进程最多timeout秒，如果一直无数据可用，抛出Empty异常（带超时的阻塞调用）。如果block为假，如果有数据可用返回数据，否则立即抛出Empty异常（非阻塞调用，timeout被忽略）。
# 如果1秒后没取到数据就退出
Queue.get(timeout = 1)

# 取数据，如果没数据抛queue.Empty异常
Queue.get_nowait()#等价于get(item, False)
# 放数据，如果满来抛queue.Full异常
Queue.put_nowait(item) #等价于put(item, False)# 返回队列的近似大小。注意，队列大小大于0并不保证接下来的get()调用不会被阻塞，队列大小小于maxsize也不保证接下来的put()调用不会被阻塞。
Queue.qsize()

# 返回True,如果空
Queue.empty() #return True if empty  

# return True if full 
Queue.full() 

# 后续调用告诉队列，任务的处理是完整的。
Queue.task_done()

join()
阻塞调用线程，直到队列中的所有任务被处理掉。只要有数据被加入队列，未完成的任务数就会增加。当消费者线程调用task_done()（意味着有消费者取得任务并完成任务），未完成的任务数就会减少。当未完成的任务数降到0，join()解除阻塞。

task_done()
意味着之前入队的一个任务已经完成。由队列的消费者线程调用。每一个get()调用得到一个任务，接下来的task_done()调用告诉队列该任务已经处理完毕。如果当前一个join()正在阻塞，它将在队列中的所有任务都处理完时恢复执行（即每一个由put()调用入队的任务都有一个对应的task_done()调用）。

exception queue.Empty

Exception raised when non-blocking get() (or get_nowait()) is called on a Queue object which is empty.

exception queue.Full

Exception raised when non-blocking put() (or put_nowait()) is called on a Queue object which is full.

 

8、生产者消费者模型

在并发编程中使用生产者和消费者模式能够解决绝大多数并发问题。该模式通过平衡生产线程和消费线程的工作能力来提高程序的整体处理数据的速度。

为什么要使用生产者和消费者模式

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

什么是生产者消费者模式

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

下面来学习一个最基本的生产者消费者模型的例子

import threading,time
import queue
q = queue.Queue(maxsize=10)
def Producer(name):
    count = 1
    while True:
        q.put("骨头%s" % count)
        print("生产了骨头",count)
        count +=1
        time.sleep(0.1)
 
def  Consumer(name):
    #while q.qsize()>0:
    while True:
        print("[%s] 取到[%s] 并且吃了它..." %(name, q.get()))
        time.sleep(1)
 
p = threading.Thread(target=Producer,args=("Alex",))
c = threading.Thread(target=Consumer,args=("ChengRonghua",))
c1 = threading.Thread(target=Consumer,args=("王森",))
p.start()
c.start()
c1.start()

进阶版：

import threading
import queue 
def producer():
    for i in range(10):
        q.put("骨头 %s" % i ) 
    print("开始等待所有的骨头被取走...")
    q.join()
    print("所有的骨头被取完了...")
 
def consumer(n):
    while q.qsize() >0: 
        print("%s 取到" %n  , q.get())
        q.task_done() #告知这个任务执行完了  
q = queue.Queue()
 
p = threading.Thread(target=producer,)
p.start() 
c1 = consumer("AJ")

 

在学习了上面线程相关，多线程的问题后，我们来看看多进程的问题

9、多进程

Unix/Linux操作系统提供了一个fork()系统调用，它非常特殊。普通的函数调用，调用一次，返回一次，但是fork()调用一次，返回两次，因为操作系统自动把当前进程（称为父进程）复制了一份（称为子进程），然后，分别在父进程和子进程内返回。

子进程永远返回0，而父进程返回子进程的ID。这样做的理由是，一个父进程可以fork出很多子进程，所以，父进程要记下每个子进程的ID，而子进程只需要调用getppid()就可以拿到父进程的ID。

Python的os模块封装了常见的系统调用，其中就包括fork，可以在Python程序中轻松创建子进程：

# -*- coding: utf-8 -*-
import os
print('Process (%s) start...' % os.getpid())
# Only works on Unix/Linux/Mac:
pid = os.fork()
if pid == 0:
    print('I am child process (%s) and my parent is %s.' % (os.getpid(), os.getppid()))
else:
    print('I (%s) just created a child process (%s).' % (os.getpid(), pid))

由于Windows没有fork调用，上面的代码在Windows上无法运行。由于Mac系统是基于BSD（Unix的一种）内核，所以，在Mac下运行是没有问题的。

有了fork调用，一个进程在接到新任务时就可以复制出一个子进程来处理新任务，常见的Apache服务器就是由父进程监听端口，每当有新的http请求时，就fork出子进程来处理新的http请求。

multiprocessing

如果你打算编写多进程的服务程序，Unix/Linux无疑是正确的选择。由于Windows没有fork调用，难道在Windows上无法用Python编写多进程的程序？

由于Python是跨平台的，multiprocessing模块就是跨平台版本的多进程模块。和threading.Thread类似。

直接从侧面用subprocesses替换线程使用GIL的方式，由于这一点，multiprocessing模块可以让程序员在给定的机器上充分的利用CPU。在multiprocessing中，通过创建Process对象生成进程，然后调用它的start()方法，

import multiprocessing
import time
def run(name):
    time.sleep(2)
    print(name, " 进程启动")
 
if __name__ == '__main__':
    mp = multiprocessing.Process(target=run, args=("LJ",))
    mp.start()
    mp.join() # 等待进程执行完毕

创建子进程时，只需要传入一个执行函数和函数的参数，创建一个Process实例，用start()方法启动，这样创建进程比fork()还要简单。

join()方法可以等待子进程结束后再继续往下运行，通常用于进程间的同步。

同样可以同时启动多个进程，进程内部可以再起线程

import multiprocessing
import time
import threading
def thread_run():
    print(threading.get_ident())
 
def run(name):
    time.sleep(2)
    print(name, " 进程启动")
    #在进程中再启动线程
    t = threading.Thread(target=thread_run, )
    t.start()
 
if __name__ == '__main__':
# 生成多个进程
    for i in range(10):
        p = multiprocessing.Process(target=run, args=('Jerey %s' %i,))
        p.start()

下面我们看看父进程与子进程的关系

我们知道Linux中每个进程都是由父进程启动的,

通过top命令可以查看Linux当前的进程情况

通过lsof -i命令查看mac当前进程清理

Linux中有一个pid为1,user为root的systemd进程，这个所早的父进程，去启动其他进程

ps -ef|awk '$2 ~ /pid/{print $3}'

命令如上通过pid查看其父进程。其中pid为已知进程pid.

__author__ = "Alex Li"
 
from multiprocessing import Process
import os
def info(title):
    print(title)
    print('module name:', __name__)
    print('parent process:', os.getppid())
    print('process id:', os.getpid())
    print("\n\n")
 
def f(name):
    info('\033[31;1mcalled from child process function f\033[0m')
    print('hello', name)
if __name__ == '__main__':
    info('\033[32;1mmain process line\033[0m')
    p = Process(target=f, args=('bob',))
    p.start()
    # p.join()

/Users/ljf/PycharmProjects/MyPython01/venv/bin/python /Users/ljf/selfDoc/pythonLearning/python-Learning/day09/getProcessId.py
main process line
module name: __main__
parent process: 2966
process id: 4034
 
 
 
called from child process function f
module name: __main__
parent process: 4034
process id: 4035
 
 
 
hello bob
 
Process finished with exit code 0

在这个结果中，mac和windows情况一样，上面程序中我们在启动主进程main后，在其中启动子进程p，通过输出的进程id及父进程pid发现，主进程main的id为4034，启动的子进程4035，那么main的父进程是2966，通过查询发现是pycharm，也就证明了每个进程都是由父进程启动的

10、进程间通信

我们知道进程之间不共享内存，如果要实现进程间通信就需要中间件，主要有以下方法：

# 管道( pipe )：管道是一种半双工的通信方式，数据只能单向流动，而且只能在具有亲缘关系的进程间使用。进程的亲缘关系通常是指父子进程关系。
# 有名管道 (named pipe) ： 有名管道也是半双工的通信方式，但是它允许无亲缘关系进程间的通信。
# 信号量( semophore ) ： 信号量是一个计数器，可以用来控制多个进程对共享资源的访问。它常作为一种锁机制，防止某进程正在访问共享资源时，其他进程也访问该资源。因此，主要作为进程间以及同一进程内不同线程之间的同步手段。
# 消息队列( message queue ) ： 消息队列是由消息的链表，存放在内核中并由消息队列标识符标识。消息队列克服了信号传递信息少、管道只能承载无格式字节流以及缓冲区大小受限等缺点。
# 信号 ( sinal ) ： 信号是一种比较复杂的通信方式，用于通知接收进程某个事件已经发生。
# 共享内存( shared memory ) ：共享内存就是映射一段能被其他进程所访问的内存，这段共享内存由一个进程创建，但多个进程都可以访问。共享内存是最快的 IPC 方式，它是针对其他进程间通信方式运行效率低而专门设计的。它往往与其他通信机制，如信号两，配合使用，来实现进程间的同步和通信。
# 套接字( socket ) ： 套解口也是一种进程间通信机制，与其他通信机制不同的是，它可用于不同及其间的进程通信。

10.1、Queues 这是一个进程的Queues

使用方法跟threading里的queue差不多

from multiprocessing import Process, Queue
def f(q):
    q.put([42, None, 'hello'])
#父进程起了进程队列q，在生成子进程p时候把这个q传给他，在父进程可以访问到子进程数据
# 看起来好像两个进程共享了一个Queue，实际上克隆了一个Queue
# 通过中间方序列化再反序列化pickle实现
if __name__ == '__main__':
    # q = queue.Queue()
    # p = threading.Thread(target=f,)
    q = Queue()
    p = Process(target=f, args=(q,))
    p.start()
    print(q.get())  # prints "[42, None, 'hello']"
    p.join()

10.2、Pipes

Pipe方法返回(conn1, conn2)代表一个管道的两个端。Pipe方法有duplex参数，如果duplex参数为True(默认值)，那么这个管道是全双工模式，也就是说conn1和conn2均可收发。duplex为False，conn1只负责接受消息，conn2只负责发送消息。

send和recv方法分别是发送和接受消息的方法。例如，在全双工模式下，可以调用conn1.send发送消息，conn1.recv接收消息。如果没有消息可接收，recv方法会一直阻塞。如果管道已经被关闭，那么recv方法会抛出EOFError。

#!/usr/bin/env python
# -*- coding: utf-8 -*-
from multiprocessing import Process, Pipe
def f(conn):
    conn.send([42, None, 'hello from child'])
    conn.send([42, None, 'hello from child3'])
    print("from parent", conn.recv())
    conn.close()
 
if __name__ == '__main__':
    parent_conn, child_conn = Pipe()  #声明一个管道的两端，默认全双工
    p = Process(target=f, args=(child_conn,))
    p.start()
    print("parent",parent_conn.recv())  # 父进程接收来自子进程p中发送的信息
    print("parent",parent_conn.recv())  # 父进程接收子进程发送的第二条消息，
    # print(parent_conn.recv()) #如果子进程没发送，那么会阻塞一直等待
    parent_conn.send(" hello child process")  # 父进程给子进程发送信息，子进程会相应接收
    p.join()

 

pipe()返回两个连接对象代表pipe的两端。每个连接对象都有send()方法和recv()方法。

但是如果两个进程或线程对象同时读取或写入管道两端的数据时，管道中的数据有可能会损坏。

当进程使用的是管道两端的不同的数据则不会有数据损坏的风险。

10.3、Managers

Python实现多进程间通信的方式如队列，管道等。只适用于多个进程都是源于同一个父进程的情况。
如果多个进程不是源于同一个父进程，只能用共享内存，信号量等方式，但是这些方式对于复杂的数据结构，

如Queue，list, dict,，Namespace， Lock， RLock，Semaphore， BoundedSemaphore， Condition， Event， Barrier， Queue， Value ， Array。 使用起来比较麻烦，不够灵活。
Manager是一种较为高级的多进程通信方式，它能支持Python支持的的任何数据结构。
它的原理是：先启动一个ManagerServer进程，这个进程是阻塞的，它监听一个socket，然后其他进程（ManagerClient）通过socket来连接到ManagerServer，实现通信。

from multiprocessing import Process, Manager
import os
def f(d, l):
    d[1] = '1'
    d['2'] = 2
    d[0.25] = None
    # d["pid%s" %os.getpid()] = os.getpid()
    l.append(os.getpid())
    print(l,d)
 
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))
            p.start()
            p_list.append(p)
        for res in p_list:
            res.join()
        l.append("from parent")
        print(d)
        print(l)

Server process manager比 shared memory 更灵活，因为它可以支持任意的对象类型。另外，一个单独的manager可以通过进程在网络上不同的计算机之间共享，不过他比shared memory要慢。　　

10.4、进程同步

进程不会共享内存，所以对进程其实无需枷锁，这里进程锁是为了保证其共享的屏幕在输出的时候不乱，比如一个进程输出在屏幕上没有完成时候，另一个进程也要输出，导致输出上一个输出信息中混入了下一个输出信息

from multiprocessing import Process, Lock
'''进程锁'''
def f(l,i):
    l.acquire()
    print('Hello World',i)
    l.release()
 
if __name__ == '__main__':
    lock = Lock()
    for num in range(10):
        Process(target=f, args=(lock,num)).start()

 

11、进程池　　

python中没有线程池的概念，由于起一个线程消耗小，可以用信号量或队列自己模拟一个线程池。

# 简单往队列中传输线程数
import threading
import time
import queue
class Threadingpool():
    def __init__(self,max_num = 10):
        self.queue = queue.Queue(max_num)
        for i in range(max_num):
            self.queue.put(threading.Thread)
    def getthreading(self):
        return self.queue.get()
    def addthreading(self):
        self.queue.put(threading.Thread)
def func(p,i):
    time.sleep(1)
    print(i)
    p.addthreading()
if __name__ == "__main__":
    p = Threadingpool()
    for i in range(20):
        thread = p.getthreading()
        t = thread(target = func, args = (p,i))
        t.start()

#往队列中无限添加任务
import queue
import threading
import contextlib
import time
 
StopEvent = object()
 
class ThreadPool(object):
 
    def __init__(self, max_num):
        self.q = queue.Queue()
        self.max_num = max_num
 
        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 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)   # 执行函数
                status = True
            except Exception as e:    # 函数执行失败
                status = False
                result = e
 
            if callback is not None:
                try:
                    callback(status, result)
                except Exception as e:
                    pass
 
            # self.free_list.append(current_thread)
            # event = self.q.get()
            # self.free_list.remove(current_thread)
            with self.work_state():
                event = self.q.get()
        else:
            self.generate_list.remove(current_thread)
 
    def close(self):
        """
        关闭线程，给传输全局非元祖的变量来进行关闭
        :return:
        """
        for i in range(len(self.generate_list)):
            self.q.put(StopEvent)
 
    def terminate(self):
        """
        突然关闭线程
        :return:
        """
        self.terminal = True
        while self.generate_list:
            self.q.put(StopEvent)
        self.q.empty()
 
    @contextlib.contextmanager
    def work_state(self):
        self.free_list.append(threading.currentThread)
        try:
            yield
        finally:
            self.free_list.remove(threading.currentThread)
 
 
def work(i):
    print(i)
    return i +1 # 返回给回调函数
 
def callback(ret):
    print(ret)
 
pool = ThreadPool(10)
for item in range(50):
    pool.run(func=work, args=(item,),callback=callback)
 
pool.terminate()
# pool.close()
 
方法二

CPU在某一时刻只能执行一个进程，那为什么上面10个进程还能够并发执行呢？实际在CPU在处理上面10个进程时是在不停的切换执行这10个进程，但由于上面10个进程的程序代码都是十分简单的，并没有涉及什么复杂的功能，并且，CPU的处理速度实在是非常快，所以这样一个过程在我们人为感知里确实是在并发执行的，实际只不过是CPU在不停地切换而已，这是通过增加切换的时间来达到目的的。     10个简单的进程可以产生这样的效果，那试想一下，如果我有100个进程需要CPU执行，但因为CPU还要进行其它工作，只能一次再处理10个进程（切换处理），否则有可能会影响其它进程工作，这下可怎么办？这时候就可以用到Python中的进程池来进行调控了，在Python中，可以定义一个进程池和这个池的大小，假如定义进程池的大小为10，那么100个进程可以分10次放进进程池中，然后CPU就可以10次并发完成这100个进程了。

进程池内部维护一个进程序列，当使用时，则去进程池中获取一个进程，如果进程池序列中没有可供使用的进进程，那么程序就会等待，直到进程池中有可用进程为止。进程池设置最好等于CPU核心数量.

构造方法：

Pool([processes[, initializer[, initargs[, maxtasksperchild[, context]]]]])

processes ：使用的工作进程的数量，如果processes是None那么使用 os.cpu_count()返回的数量。
initializer： 如果initializer是None，那么每一个工作进程在开始的时候会调用initializer(*initargs)。
maxtasksperchild：工作进程退出之前可以完成的任务数，完成后用一个新的工作进程来替代原进程，来让闲置的资源被释放。maxtasksperchild默认是None，意味着只要Pool存在工作进程就会一直存活。
context: 用在制定工作进程启动时的上下文，一般使用 multiprocessing.Pool() 或者一个context对象的Pool()方法来创建一个池，两种方法都适当的设置了context

方法:

• apply(func[, args[, kwds]]) ：使用arg和kwds参数调用func函数，结果返回前会一直阻塞，由于这个原因，apply_async()更适合并发执行，另外，func函数仅被pool中的一个进程运行。

• apply_async(func[, args[, kwds[, callback[, error_callback]]]]) ： apply()方法的一个变体，会返回一个结果对象。如果callback被指定，那么callback可以接收一个参数然后被调用，当结果准备好回调时会调用callback，调用失败时，则用error_callback替换callback。 Callbacks应被立即完成，否则处理结果的线程会被阻塞。主进程调用该回掉函数。

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

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

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

from  multiprocessing import Process, Pool,freeze_support
import time,os
def Foo(i):
    time.sleep(2)
    print("in process",os.getpid())
    return i + 100
def Bar(arg):
    #该回掉函数比如访问数据库结束后，往log中写日志，也可以子进程做这个操作，但是放到这里，只需要父进程连接数据库一次
    #然后每次子进程结束后，父进程回掉写日志，避免子进程多次访问数据库
    print('-->exec done:', arg,os.getpid())
if __name__ == '__main__':
    # freeze_support() #如果在windows环境上启动多进程需要加这一句
    pool = Pool(processes=3) #允许进程池同时放入5个进程
    print("主进程",os.getpid())
    for i in range(10):
        # pool.apply_async(func=Foo, args=(i,), callback=Bar) #callback=回调
        # pool.apply(func=Foo, args=(i,)) #串行
        pool.apply_async(func=Foo, args=(i,)) #并行
    print('end')
    pool.close()
    pool.join() #进程池中进程执行完毕后再关闭，如果注释，那么程序直接关闭。.join()

使用apply()方法时候，实际上是串行的，使用apply_async()会是真正并行，代码中当设置进程池Pool进程数为3，运行代码发现进程启动时候每3个启动。

多核CPU

如果你不幸拥有一个多核CPU，你肯定在想，多核应该可以同时执行多个线程。

如果写一个死循环的话，会出现什么情况呢？

打开Mac OS X的Activity Monitor，或者Windows的Task Manager，都可以监控某个进程的CPU使用率。

我们可以监控到一个死循环线程会100%占用一个CPU。

如果有两个死循环线程，在多核CPU中，可以监控到会占用200%的CPU，也就是占用两个CPU核心。

要想把N核CPU的核心全部跑满，就必须启动N个死循环线程。

用Python写个死循环：

启动与CPU核心数量相同的N个线程，在4核CPU上可以监控到CPU占用率仅有102%，也就是仅使用了一核。

但是用C、C++或Java来改写相同的死循环，直接可以把全部核心跑满，4核就跑到400%，8核就跑到800%，为什么Python不行呢？

因为Python的线程虽然是真正的线程，但解释器执行代码时，有一个GIL锁：Global Interpreter Lock，任何Python线程执行前，必须先获得GIL锁，然后，每执行100条字节码，解释器就自动释放GIL锁，让别的线程有机会执行。这个GIL全局锁实际上把所有线程的执行代码都给上了锁，所以，多线程在Python中只能交替执行，即使100个线程跑在100核CPU上，也只能用到1个核。

GIL是Python解释器设计的历史遗留问题，通常我们用的解释器是官方实现的CPython，要真正利用多核，除非重写一个不带GIL的解释器。

所以，在Python中，可以使用多线程，但不要指望能有效利用多核。如果一定要通过多线程利用多核，那只能通过C扩展来实现，不过这样就失去了Python简单易用的特点。

不过，也不用过于担心，Python虽然不能利用多线程实现多核任务，但可以通过多进程实现多核任务。多个Python进程有各自独立的GIL锁，互不影响。