必备的理论基础

1.操作系统作用：

• 隐藏丑陋复杂的硬件接口，提供良好的抽象接口。
• 管理调度进程，并将多个进程对硬件的竞争变得有序
  2.多道技术：
• 针对单核，实现并发
  • 现在的计算机是一般是多核，但是每个核都会用到多道技术，有4个cpu就，运行于cpu1的进程遇到io阻塞，就会等到io结束后重行调度，会被调度到4个cpu中的任何一个，由操作系统调度算法决定。
• 空间上的复用：在内存中同时运行多道程序。
• 时间上复用：复用一个cpu时间片
  • 强调：遇到io切，占用cpu时间过长也切，核心在于切之前将进程的状态保存下来，这样才能保证下次切回来时，能基于上次切走的位置继续运行。

并发编程之多线程

进程理论

进程
正在进行的一个过程或者说是一个任务。而负责任务则是CPU。

进程与程序的区别
程序仅仅是一堆代码，而进程指的是程序的运行过程。
同一个程序执行两次，那也是两个进程。

并发与并行
无论是并行还是并发，在用户看来都是‘同时’运行的，不管是进程还是线程，都只是一个任务而已，真正干活的是cpu，cpu来做这些任务，而一个cpu同一时刻只能执行一个任务

1. 并发： 是伪并行，即看起来是同时运行。单个cpu+多道技术就可以实现并发。
2. 并行：同时运行，只有具备多个cpu才能实现并行

进程的创建
对于通用系统，需要有系统运行过程中创建或撤销进程的能力，主要分为4中形式创建新的进程：

1. 系统初始化
2. 一个进程在运行过程中开启了子进程
3. 用户的交互式请求，而创建以个新进程
4. 一个批处理作业的初始化
   无论是哪一种，新进程的创建都是由一个已经存在的进程执行了一个用于创建进程的系统调用而创建：
5. 在UNIX中该系统调用是：fork，fork会创建一个与父进程一模一样的副本，二者有相同的存储映像、同样的环境字符串和同样的打开文件；
6. 在windows中该系统调用是： CreateProcess,CreateProcess即处理进程的创建，也负责把正确的程序装入新进程。

关于创建的子进程，UNIX和windows：

1. 相同的是：进程创建后，父进程和子进程有各自不同的地址空间（多道技术要切物理层面实现进程之间的内存隔离）， 任何一个进程在其地址空间中的修改都不会影响到另外一个进程。
2. 不同的是：在unix中，子进程的初始地址空间是父进程的一个副本，子进程和父进程是可以有只读的共享内存区的；windows系统来说，从一开始父进程与子进程地址空间就是不同的。

进程的终止

1. 正常退出
2. 出错退出
3. 严重错误
4. 被其他进程杀死

进程的层次结构
无论是UNIX还是windows，进程只有一个父进程，不同的是：

1. 在unix中所有的进程，都是以init进程为根，组成树形结构。父子进程共同组成一个进程组，这样，当从键盘发出一个信号时，该信号被送给当前与键盘相关的进程 组中的所有成员。
2. 在windows中，没有进程层次的概念，所有的进程都是地位相同的，唯一类似于进程层次的韩式，是在创建进程时，父进程得到一个特别的令牌（称之为句柄），该句柄可以用来控制进程，但是父进程有权把该句柄传给其他子进程，这样就没有层次了。

进程的状态
一个进程有如下三种状态：

进程的三种状状态.png

进程并发的实现
进程并发的实现在于，硬件中断一个正在运行的进程，把此时进程运行的所有状态表寸下来，为此，操作系统维护一张表格，即进程表（process table），每个进程占用一个进程表项（这些表项也称为进程控制块）

进程表.png

开启进程的两种方式

multiprocessing模块介绍
python中的多线程无法利用多核优势，如果想要充分的使用多核CPU的资源（os.cpu_count()查看），在python中大部分情况需要时用多进程。
python中提供了multiprocessing。multiprocessing模块用来开启子进程，并在子进程中执行我们定制的任务，该模块与多线程模块threading的编程接口类似。multiprocessing模块的功能众多： 支持子进程、通信和共享数据、执行不同形式的同步， 提供了Process、Queue、Pipe、Lock等组件。
需要强调的一点是：与线程不同, 进程没有任何共享状态，进程修改的数据，改动仅限于该进程内。

Process类介绍

Process([group [, target [, name [, args [, kwargs]]]]])，由该类实例化得到的对象，可用来开启一个子进程

强调：
1. 需要使用关键字的方式来指定参数
2. args指定的为传给target函数的位置参数，是一个元组形式，必须有逗号

方法介绍：

p.start()：启动进程，并调用该子进程中的p.run() 
p.run():进程启动时运行的方法，正是它去调用target指定的函数，我们自定义类的类中一定要实现该方法  

p.terminate():强制终止进程p，不会进行任何清理操作，如果p创建了子进程，该子进程就成了僵尸进程，使用该方法需要特别小心这种情况。如果p还保存了一个锁那么也将不会被释放，进而导致死锁
p.is_alive():如果p仍然运行，返回True

p.join([timeout]):主线程等待p终止（强调：是主线程处于等的状态，而p是处于运行的状态）。timeout是可选的超时时间。

属性介绍：

p.daemon：默认值为False，如果设为True，代表p为后台运行的守护进程，当p的父进程终止时，p也随之终止，并且设定为True后，p不能创建自己的新进程，必须在p.start()之前设置

p.name:进程的名称

p.pid：进程的pid

Process类的使用
注意：在windows中Process()必须放到# if __name__ == '__main__':下
创建并开启子进程的方式一:

#!/usr/bin/env python3
# Author    : fbo
# @Time : 18-5-19 上午7:27
# @File : 01 开启进程的两种方式.py
import time
import random
from multiprocessing import Process

def piao(name):
    print('%s piaoing' % name)
    time.sleep(random.randint(1, 5))
    print('%s piao end' % name)

if __name__ == '__main__':
    p1 = Process(target=piao,args=('alex',))    # 必须加逗号
    p2 = Process(target=piao,args=('egon',))
    p3 = Process(target=piao,args=('wupeiqi',))
    p4 = Process(target=piao,args=('yuanhao',))

    # 调用对象下的方法,启动四个进程
    p1.start()
    p2.start()
    p3.start()
    p4.start()

    print("主进程")

创建并开启子进程的方式二:

import time, random
from multiprocessing import Process

class Piao(Process):
    def __init__(self, name):
        super().__init__()
        self.name = name

    def run(self):
        print('%s is piaoing' % self.name)
        time.sleep(random.randint(1, 5))
        print('%s piao end' % self.name)

if __name__ == '__main__':
    p1 = Piao('egon')
    p2 = Piao('Alex')
    p3 = Piao('wupeiqi')
    p4 = Piao('yuanhao')

    p1.start()
    p2.start()
    p3.start()
    p4.start()

    print('主进程')

练习题

1. 思考开启进程的方式一和方式二各开启了几个进程？
   都有5个进程，一个主进程，四个子进程
2. 进程之间的内存空间是共享的还是隔离的？下述代码的执行结果是什么？

from multiprocessing import Process

n=100 #在windows系统中应该把全局变量定义在if __name__ == '__main__'之上就可以了

def work():
    global n
    n=0
    print('子进程内: ',n)


if __name__ == '__main__':
    p=Process(target=work)
    p.start()
    print('主进程内: ',n)

主进程内: 100
子进程内: 0

3. 基于多进程实现并发的套接字通信？

#!/usr/bin/env python3
# Author    : fbo
# @Time : 18-5-19 下午8:30
# @File : 服务端.py

import socket
import os
from multiprocessing import Process

def handle_data(conn, addr):
    print('Pid >>> ', os.getpid())
    print('connect by ', addr)
    while True:
        data = conn.recv(1024)
        if not data: break
        conn.send(data.upper())
    conn.close()

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind(('127.0.0.1', 8080))
sock.listen(10)
print('Server starting...')

while True:
    conn, addr = sock.accept()
    p = Process(target=handle_data, args=(conn,addr,))
    p.start()

#!/usr/bin/env python3
# Author    : fbo
# @Time : 18-5-19 下午8:30
# @File : 客户端.py

import socket

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

sock.connect(('127.0.0.1', 8080))

while True:
    data = input('>>: ').strip()
    if not data: continue
    sock.send(data.encode('utf-8'))
    rt = sock.recv(1024).decode('utf-8')
    print(rt)

4. 思考每来一个客户端，服务端就开启一个新的进程来服务它，这种实现方式有无问题？

这种实现方式会造成占用过多的系统资源。

Process对象的join方法

在主进程运行过程中如果想并发地执行其他的任务，我们可以开启子进程，此时主进程的任务与子进程的任务分两种情况

情况一：在主进程的任务与子进程的任务彼此独立的情况下，主进程的任务先执行完毕后，主进程还需要等待子进程执行完毕，然后统一回收资源。

情况二：如果主进程的任务在执行到某一个阶段时，需要等待子进程执行完毕后才能继续执行，就需要有一种机制能够让主进程检测子进程是否运行完毕，在子进程执行完毕后才继续执行，否则一直在原地阻塞，这就是join方法的作用

#!/usr/bin/env python3
# Author    : fbo
# @Time : 18-5-19 下午9:53
# @File : 02 Process join.py

from multiprocessing import Process
import time
import os
import random

def task():
    print('%s is piaoing...' % os.getpid())
    time.sleep(random.randint(1,5))
    print('%s piao end' % os.getpid())

if __name__ == '__main__':
    p1 = Process(target=task, args=())

    p1.start()
    p1.join() # 等p1完成后才执行后续代码
    print('main process...')

有了join，程序不就是串行了吗？？？

from multiprocessing import Process
import time
import os
import random

def task():
    print('%s is piaoing...' % os.getpid())
    time.sleep(random.randint(1,5))
    print('%s piao end' % os.getpid())

if __name__ == '__main__':
    p1 = Process(target=task, args=())
    p2 = Process(target=task, args=())
    p3 = Process(target=task, args=())
    p4 = Process(target=task, args=())

    p1.start()
    p2.start()
    p3.start()
    p4.start()
    # 使用ｊｏｉｎ方法后并不意味着变成了串行了，只是主进程需要等子进程运行完成，
    # 各个子进程还是并行运行
    p1.join()
    p2.join()
    p3.join()
    p4.join()
    print('main process...')

Process对象的其他属性或方法

from multiprocessing import Process
import time
import random

def task(name):
    print('%s is piaoing' %name)
    time.sleep(random.randrange(1,5))
    print('%s is piao end' %name)

if __name__ == '__main__':
    p1=Process(target=task,args=('egon',))
    p1.start()

    p1.terminate()#关闭进程,不会立即关闭,所以is_alive立刻查看的结果可能还是存活
    print(p1.is_alive()) #结果为True

    print('主')
    print(p1.is_alive()) #结果为False

from multiprocessing import Process
import time
import random

def task(name):
    print('%s is piaoing' %name)
    time.sleep(random.randrange(1,5))
    print('%s is piao end' %name)

if __name__ == '__main__':
    # p1=Process(target=task,args=('egon',),name='子进程1') #可以用关键参数来指定进程名
    p1=Process(target=task,args=('egon',)) # name='子进程1') #可以用关键参数来指定进程名
    p1.start()

    print(p1.name,p1.pid,)

守护进程

主进程创建子进程，然后将该进程设置成守护自己的进程.

• 守护进程会在主进程代码执行结束后就终止
• 守护进程内无法再开启子进程,否则抛出异常：AssertionError: daemonic processes are not allowed to have children

如果我们有两个任务需要并发执行，那么开一个主进程和一个子进程分别去执行就ok了，如果子进程的任务在主进程任务结束后就没有存在的必要了，那么该子进程应该在开启前就被设置成守护进程。主进程代码运行结束，守护进程随即终止

from multiprocessing import Process
import time
import random

def task(name):
    print('%s is piaoing' %name)
    time.sleep(random.randrange(1,3))
    print('%s is piao end' %name)


if __name__ == '__main__':
    p=Process(target=task,args=('egon',))
    p.daemon=True #一定要在p.start()前设置,设置p为守护进程,禁止p创建子进程,并且父进程代码执行结束,p即终止运行
    p.start()
    print('主') #只要终端打印出这一行内容，那么守护进程p也就跟着结束掉了

练习题
思考下列代码的执行结果有可能有哪些情况？为什么？

from multiprocessing import Process

import time
def foo():
    print(123)
    time.sleep(1)
    print("end123")

def bar():
    print(456)
    time.sleep(3)
    print("end456")

if __name__ == '__main__':
    p1=Process(target=foo)
    p2=Process(target=bar)

    p1.daemon=True
    p1.start()
    p2.start()
    print("main-------")

main------
456
end456

互斥锁

进程之间数据不共享,但是共享同一套文件系统,所以访问同一个文件,或同一个打印终端,是没有问题的,而共享带来的是竞争，竞争带来的结果就是错乱，如下：

#!/usr/bin/env python3
# Author    : fbo
# @Time : 18-5-20 下午8:03
# @File : 03互斥锁.py

from multiprocessing import Process,Lock
import os, time

'''
由串行变成并行,避免了资源争抢，但是牺牲了运行效率
'''
def task(lock):
    lock.acquire()
    print(os.getpid(), 'is running...')
    time.sleep(2)
    print(os.getpid(), ' ending...')
    lock.release()


if __name__ == '__main__':
    lock = Lock()
    for i in range(3):
        p = Process(target=task, args=(lock,))
        p.start()

模拟抢票练习
多个进程共享同一文件，我们可以把文件当数据库，用多个进程模拟多个人执行抢票任务

#!/usr/bin/env python3
# Author    : fbo
# @Time : 18-5-20 下午8:24
# @File : 03 模拟抢票软件.py

import json, time
from multiprocessing import Process

def search(name):
    dic = json.load(open('db.json'))
    time.sleep(1)
    print('\033[43m %s 查到剩余票数%s\033[0m' %(name, dic['count']))


def get(name):
    dic = json.load(open('db.json'))
    time.sleep(1)
    if dic['count'] > 0:
        dic['count'] -= 1
        time.sleep(1)
        json.dump(dic, open('db.json', 'w'))
        print('\033[46m%s购票成功\033[0m' % name)

def task(name):
    search(name)
    get(name)


if __name__ == '__main__':
    for i in range(10):
        name = '<路人%s>' % i
        p = Process(target=task, args=(name,))
        p.start()
 '''
 运行结果：       
 <路人0> 查到剩余票数1
 <路人1> 查到剩余票数1
 <路人3> 查到剩余票数1
 <路人4> 查到剩余票数1
 <路人2> 查到剩余票数1
 <路人5> 查到剩余票数1
 <路人7> 查到剩余票数1
 <路人6> 查到剩余票数1
 <路人8> 查到剩余票数1
 <路人9> 查到剩余票数1
<路人0>购票成功
<路人1>购票成功
<路人3>购票成功
<路人4>购票成功
<路人2>购票成功
<路人5>购票成功
<路人7>购票成功
<路人6>购票成功
<路人8>购票成功
<路人9>购票成功
'''

并发运行，效率高，但竞争写同一文件，数据写入错乱,只有一张票，卖成功给了10个人

加锁处理：购票行为由并发变成了串行，牺牲了运行效率，但保证了数据安全

import json, time
from multiprocessing import Process,Lock

def search(name):
    dic = json.load(open('db.json'))
    time.sleep(1)
    print('\033[43m %s 查到剩余票数%s\033[0m' %(name, dic['count']))


def get(name):
    dic = json.load(open('db.json'))
    time.sleep(1)
    if dic['count'] > 0:
        dic['count'] -= 1
        time.sleep(1)
        json.dump(dic, open('db.json', 'w'))
        print('\033[46m%s购票成功\033[0m' % name)

def task(name, lcok):
    search(name)
    with lock: #相当于lock.acquire(),执行完自代码块自动执行lock.release()
        get(name)


if __name__ == '__main__':
    lock = Lock()
    for i in range(10):
        name = '<路人%s>' % i
        p = Process(target=task, args=(name,lock))
        p.start()
        
'''
 <路人1> 查到剩余票数1
 <路人0> 查到剩余票数1
 <路人2> 查到剩余票数1
 <路人3> 查到剩余票数1
 <路人5> 查到剩余票数1
 <路人4> 查到剩余票数1
 <路人6> 查到剩余票数1
 <路人7> 查到剩余票数1
 <路人9> 查到剩余票数1
 <路人8> 查到剩余票数1
<路人1>购票成功
'''

互斥锁与join
使用join可以将并发变成串行，互斥锁的原理也是将并发变成穿行，那我们直接使用join就可以了啊，为何还要互斥锁，说到这里我赶紧试了一下

import json, time
from multiprocessing import Process,Lock

def search(name):
    dic = json.load(open('db.json'))
    time.sleep(1)
    print('\033[43m %s 查到剩余票数%s\033[0m' %(name, dic['count']))


def get(name):
    dic = json.load(open('db.json'))
    time.sleep(1)
    if dic['count'] > 0:
        dic['count'] -= 1
        time.sleep(1)
        json.dump(dic, open('db.json', 'w'))
        print('\033[46m%s购票成功\033[0m' % name)

def task(name):
    search(name)
    get(name)


if __name__ == '__main__':
    for i in range(10):
        name = '<路人%s>' % i
        p = Process(target=task, args=(name,))
        p.start()
        p.join()
        
'''
 <路人0> 查到剩余票数1
<路人0>购票成功
 <路人1> 查到剩余票数0
 <路人2> 查到剩余票数0
 <路人3> 查到剩余票数0
 <路人4> 查到剩余票数0
 <路人5> 查到剩余票数0
 <路人6> 查到剩余票数0
 <路人7> 查到剩余票数0
 <路人8> 查到剩余票数0
 <路人9> 查到剩余票数0
 '''

发现使用join将并发改成穿行，确实能保证数据安全，但问题是连查票操作也变成只能一个一个人去查了，很明显大家查票时应该是并发地去查询而无需考虑数据准确与否，此时join与互斥锁的区别就显而易见了，join是将一个任务整体串行，而互斥锁的好处则是可以将一个任务中的某一段代码串行，比如只让task函数中的get任务串行

def task(name,):
    search(name) # 并发执行

    lock.acquire()
    get(name) #串行执行
    lock.release()

总结
加锁可以保证多个进程修改同一块数据时，同一时间只能有一个任务可以进行修改，即串行地修改，没错，速度是慢了，但牺牲了速度却保证了数据安全。
虽然可以用文件共享数据实现进程间通信，但问题是：

1. 效率低（共享数据基于文件，而文件是硬盘上的数据）
2. 需要自己加锁处理
   这就是mutiprocessing模块为我们提供的基于消息的IPC通信机制：队列和管道。

队列和管道都是将数据存放于内存中，而队列又是基于（管道+锁）实现的，可以让我们从复杂的锁问题中解脱出来，因而队列才是进程间通信的最佳选择。

我们应该尽量避免使用共享数据，尽可能使用消息传递和队列，避免处理复杂的同步和锁问题，而且在进程数目增多时，往往可以获得更好的可获展性。

队列

队列介绍
进程彼此之间相互隔离，要实现进程间通信（IPC），multiprocessing模块支持两种形式：队列和管道，这两种方式都是使用消息传递的。
创建队列的类（底层就是以管道和锁定的方式实现）：
Queue([maxsize]): 创建共享的进程队列，Queue是多进程安全的队列，可以使用Queue实现多进程之间的数据传递。
参数介绍：

maxsize是队列中允许最大项数，省略则无大小限制。
但是要明确：
1. 队列存放的是消息而非大数据。
2. 队列占用的是内存空间，因而maxsize即便是无大小限制也受限于内存大小。

主要方法介绍：

q.put方法用以插入数据队列中。
q.get方法可以从队列读取并且删除一个元素。

队列的使用

#!/usr/bin/env python3
# Author    : fbo
# @Time : 18-5-20 下午9:26
# @File : 04 队列.py

from multiprocessing import Process, Queue

q = Queue(3)

# put, get , put_nowait, get_nowait, full, empty
q.put(1)
q.put(2)
q.put(3)
print(q.full()) # 满了
# q.put(4) # 再放就阻塞了

print(q.get())
print(q.get())
print(q.get())
print(q.empty()) # 空了

# print(q.get())　＃　再取就阻塞了

生产者与消费者模型

生产者消费者模型介绍

为什么要使用生产者消费者模型
生产者指的是生产数据的任务，消费者指的是处理数据的任务，在并发编程中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。
什么是生产者和消费者模式
生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。

这个阻塞队列就是用来给生产者和消费者解耦的

基于上一小节学习的队列来实习一个生产者消费者模型：

#!/usr/bin/env python3
# Author    : fbo
# @Time : 18-5-20 下午9:37
# @File : 06 生产者消费者模式.py

from multiprocessing import Process, Queue
import time, random, os

def consumer(q, name):
    while True:
        res = q.get()
        time.sleep(random.randint(1,3))
        print('\033[043m%s 吃 %s\033[0m' % (name,res))

def producer(q, name, food):
    for i in range(3):
        time.sleep(random.randint(1,3))
        res = '%s%s' % (food, i)
        q.put(res)
        print('\033[45m%s 生产了　%s\033[0m' % (name, res))


if __name__ == '__main__':
    q = Queue()

    p1 = Process(target=producer, args=(q,'fbo', 'baozi ',))

    c1 = Process(target=consumer, args=(q,'alex',))

    p1.start()
    c1.start()
    print('main')

此时的问题是主进程永远不会结束，原因是：生产者p在生产完后就结束了，但是消费者c在取空了q之后，则一直处于死循环中且卡在q.get()这一步。

解决方式无非是让生产者在生产完毕后，往队列中再发一个结束信号，这样消费者在接收到结束信号后就可以break出死循环

from multiprocessing import Process,Queue
import time,random,os
def consumer(q,name):
    while True:
        res=q.get()
        if res is None:break
        time.sleep(random.randint(1,3))
        print('\033[43m%s 吃 %s\033[0m' %(name,res))

def producer(q,name,food):
    for i in range(3):
        time.sleep(random.randint(1,3))
        res='%s%s' %(food,i)
        q.put(res)
        print('\033[45m%s 生产了 %s\033[0m' %(name,res))

if __name__ == '__main__':
    q=Queue()
    #生产者们:即厨师们
    p1=Process(target=producer,args=(q,'egon1','包子'))
    p2=Process(target=producer,args=(q,'egon2','骨头'))
    p3=Process(target=producer,args=(q,'egon3','泔水'))

    #消费者们:即吃货们
    c1=Process(target=consumer,args=(q,'alex1'))
    c2=Process(target=consumer,args=(q,'alex2'))

    #开始
    p1.start()
    p2.start()
    p3.start()
    c1.start()
    c2.start()

    p1.join()
    p2.join()
    p3.join()
    q.put(None)
    q.put(None)
    q.put(None)
    print('主')

但上述解决方式，在有多个生产者和多个消费者时，我们则需要用一个很low的方式去解决,有几个消费者就需要发送几次结束信号
有另外一种队列提供了这种机制,JoinableQueue([maxsize])
参数介绍
maxsize是队列中允许最大项数，省略则无大小限制。
方法介绍:

JoinableQueue的实例p除了与Queue对象相同的方法之外还具有：
q.task_done()：使用者使用此方法发出信号，表示q.get()的返回项目已经被处理。如果调用此方法的次数大于从队列中删除项目的数量，将引发ValueError异常
q.join():生产者调用此方法进行阻塞，直到队列中所有的项目均被处理。阻塞将持续到队列中的每个项目均调用q.task_done（）方法为止

基于JoinableQueue实现生产者消费者模型

from multiprocessing import Process,JoinableQueue
import time,random,os
def consumer(q,name):
    while True:
        res=q.get()
        time.sleep(random.randint(1,3))
        print('\033[43m%s 吃 %s\033[0m' %(name,res))
        q.task_done() #发送信号给q.join()，说明已经从队列中取走一个数据并处理完毕了

def producer(q,name,food):
    for i in range(3):
        time.sleep(random.randint(1,3))
        res='%s%s' %(food,i)
        q.put(res)
        print('\033[45m%s 生产了 %s\033[0m' %(name,res))
    q.join() #等到消费者把自己放入队列中的所有的数据都取走之后，生产者才结束

if __name__ == '__main__':
    q=JoinableQueue() #使用JoinableQueue()

    #生产者们:即厨师们
    p1=Process(target=producer,args=(q,'egon1','包子'))
    p2=Process(target=producer,args=(q,'egon2','骨头'))
    p3=Process(target=producer,args=(q,'egon3','泔水'))

    #消费者们:即吃货们
    c1=Process(target=consumer,args=(q,'alex1'))
    c2=Process(target=consumer,args=(q,'alex2'))
    c1.daemon=True
    c2.daemon=True

    #开始
    p1.start()
    p2.start()
    p3.start()
    c1.start()
    c2.start()

    p1.join()
    p2.join()
    p3.join()
    #1、主进程等生产者p1、p2、p3结束
    #2、而p1、p2、p3是在消费者把所有数据都取干净之后才会结束
    #3、所以一旦p1、p2、p3结束了，证明消费者也没必要存在了，应该随着主进程一块死掉，因而需要将生产者们设置成守护进程
    print('主')

生产者消费者模型总结

1. 程序中有两类角色

• 一类负责生产数据（生产者）
• 一类负责处理数据（消费者）

2. 引入生产者消费者模型为了解决的问题是

• 平衡生产者与消费者之间的速度差
• 程序解开耦合

3. 如何实现生产者消费者模型

• 生产者<--->队列<--->消费者

并发编程之多线程

线程理论

什么是线程
在传统操作系统中，每个进程有一个地址空间，而且默认就有一个控制线程

线程顾名思义，就是一条流水线工作的过程（流水线的工作需要电源，电源就相当于cpu），而一条流水线必须属于一个车间，一个车间的工作过程是一个进程，车间负责把资源整合到一起，是一个资源单位，而一个车间内至少有一条流水线。

所以，进程只是用来把资源集中到一起（进程只是一个资源单位，或者说资源集合），而线程才是cpu上的执行单位。

多线程（即多个控制线程）的概念是，在一个进程中存在多个线程，多个线程共享该进程的地址空间，相当于一个车间内有多条流水线，都共用一个车间的资源。例如，北京地铁与上海地铁是不同的进程，而北京地铁里的13号线是一个线程，北京地铁所有的线路共享北京地铁所有的资源，比如所有的乘客可以被所有线路拉。

线程与进程的区别

1. Threads share the address space of the process that created it; processes have their own address space.
2. Threads have direct access to the data segment of its process; processes have their own copy of the data segment of the parent process.
3. Threads can directly communicate with other threads of its process; processes must use interprocess communication to communicate with sibling processes.
4. New threads are easily created; new processes require duplication of the parent process.
5. Threads can exercise considerable control over threads of the same process; processes can only exercise control over child processes.
6. Changes to the main thread (cancellation, priority change, etc.) may affect the behavior of the other threads of the process; changes to the parent process does not affect child processes.

总结上述区别，无非两个关键点，这也是我们在特定的场景下需要使用多线程的原因：

1. 同一个进程内的多个线程共享该进程内的地址资源
2. 创建线程的开销要远小于创建进程的开销（创建一个进程，就是创建一个车间，涉及到申请空间，而且在该空间内建至少一条流水线，但创建线程，就只是在一个车间内造一条流水线，无需申请空间，所以创建开销小）

开启线程的两种方式

threading模块介绍
multiprocess模块的完全模仿了threading模块的接口，二者在使用层面，有很大的相似性，因而不再详细介绍
开启线程的两种方式

#!/usr/bin/env python3
# Author    : fbo
# @Time : 18-5-21 下午9:18
# @File : 07 开启线程的两种方式.py

from threading import Thread
import time

def sayhi(name):
    time.sleep(2)
    print('%s say hello' % name)

if __name__ == '__main__':
    t = Thread(target=sayhi, args=('fbo',))
    t.start()
    print('main thread')

from threading import Thread
import time

class Sayhi(Thread):
    def __init__(self, name):
        super().__init__()
        self.name = name

    def run(self):
        time.sleep(2)
        print('%s say hello' % self.name)

if __name__ == '__main__':
    t = Sayhi('fbo')
    t.start()
    print('main threading')

练习题

1. 基于多线程实现并发的套接字通信

#!/usr/bin/env python3
# Author    : fbo
# @Time : 18-5-19 下午8:30
# @File : 服务端.py

import socket
import os
from threading import Thread

def handle_data(conn, addr):
    print('Pid >>> ', os.getpid())
    print('connect by ', addr)
    while True:
        data = conn.recv(1024)
        if not data: break
        conn.send(data.upper())
    conn.close()

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind(('127.0.0.1', 8080))
sock.listen(10)
print('Server starting...')

while True:
    conn, addr = sock.accept()
    p = Thread(target=handle_data, args=(conn,addr,))
    p.start()

#!/usr/bin/env python3
# Author    : fbo
# @Time : 18-5-19 下午8:30
# @File : 客户端.py

import socket

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

sock.connect(('127.0.0.1', 8080))

while True:
    data = input('>>: ').strip()
    if not data: continue
    sock.send(data.encode('utf-8'))
    rt = sock.recv(1024).decode('utf-8')
    print(rt)

2. 编写一个简单的文本处理工具，具备三个任务，一个接收用户输入，一个将用户输入的内容格式化成大写，一个将格式化后的结果存入文件

from threading import Thread

msg_l = []
format_l = []


def talk():
    while True:
        msg = input(">>>").strip()
        if not msg:
            break
        msg_l.append(msg)


def format_msg():
    while True:
        if msg_l:
            res = msg_l.pop()
            format_l.append(res.upper())


def save():
    while True:
        if format_l:
            with open('db.txt', 'a', encoding='utf-8') as f:
                res = format_l.pop()
                print(res)
                f.write('%s\n' % res)


if __name__ == '__main__':
    t1 = Thread(target=talk)
    t2 = Thread(target=format_msg)
    t3 = Thread(target=save)

    t1.start()
    t2.start()
    t3.start()

多线程与多进程的区别

几乎是t.start ()的同时就将线程开启了，然后先打印出了hello，证明线程的创建开销极小

p.start ()将开启进程的信号发给操作系统后，操作系统要申请内存空间，让好拷贝父进程地址空间到子进程，开销远大于线程

在主进程下开启多个线程,每个线程都跟主进程的pid一样

开多个进程,每个进程都有不同的pid

进程之间地址空间是隔离的

同一进程内开启的多个线程是共享该进程地址空间的

Thread对象的其他属性或方法

介绍

Thread实例对象的方法
  # isAlive(): 返回线程是否活动的。
  # getName(): 返回线程名。
  # setName(): 设置线程名。

threading模块提供的一些方法：
  # threading.currentThread(): 返回当前的线程变量。
  # threading.enumerate(): 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前，不包括启动前和终止后的线程。
  # threading.activeCount(): 返回正在运行的线程数量，与len(threading.enumerate())有相同的结果。

验证

from threading import Thread
import threading
from multiprocessing import Process
import os

def work():
    import time
    time.sleep(3)
    print(threading.current_thread().getName())


if __name__ == '__main__':
    #在主进程下开启线程
    t=Thread(target=work)
    t.start()

    print(threading.current_thread().getName())
    print(threading.current_thread()) #主线程
    print(threading.enumerate()) #连同主线程在内有两个运行的线程
    print(threading.active_count())
    print('主线程/主进程')

'''
MainThread
<_MainThread(MainThread, started 140735268892672)>
[<_MainThread(MainThread, started 140735268892672)>, <Thread(Thread-1, started 123145307557888)>]
主线程/主进程
Thread-1
'''

主线程等待子线程结束

from threading import Thread
import time
def sayhi(name):
    time.sleep(2)
    print('%s say hello' %name)

if __name__ == '__main__':
    t=Thread(target=sayhi,args=('egon',))
    t.start()
    t.join()
    print('主线程')
    print(t.is_alive())

'''
egon say hello
主线程
False
'''

守护线程

无论是进程还是线程，都遵循：守护xxx会等待主xxx运行完毕后被销毁

需要强调的是：运行完毕并非终止运行

1、对主进程来说，运行完毕指的是主进程代码运行完毕

2、对主线程来说，运行完毕指的是主线程所在的进程内所有非守护线程统统运行完毕，主线程才算运行完毕

1、主进程在其代码结束后就已经算运行完毕了（守护进程在此时就被回收）,然后主进程会一直等非守护的子进程都运行完毕后回收子进程的资源(否则会产生僵尸进程)，才会结束，

2、主线程在其他非守护线程运行完毕后才算运行完毕（守护线程在此时就被回收）。因为主线程的结束意味着进程的结束，进程整体的资源都将被回收，而进程必须保证非守护线程都运行完毕后才能结束。

实例：

from threading import Thread
import time
def sayhi(name):
    time.sleep(2)
    print('%s say hello' %name)

if __name__ == '__main__':
    t=Thread(target=sayhi,args=('egon',))
    t.setDaemon(True) #必须在t.start()之前设置
    t.start()

    print('主线程')
    print(t.is_alive())

'''
主线程
True
'''

练习
思考下述代码的执行结果有可能是哪些情况？为什么？

from threading import Thread
import time

def foo():
    print(123)
    time.sleep(1)
    print("end123")

def bar():
    print(456)
    time.sleep(3)
    print("end456")

if __name__ == '__main__':
    t1=Thread(target=foo)
    t2=Thread(target=bar)

    t1.daemon=True
    t1.start()
    t2.start()
    print("main-------")

GIL全局解析器锁

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.)
首先需要明确的一点是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介绍
GIL本质就是一把互斥锁，既然是互斥锁，所有互斥锁的本质都一样，都是将并发运行变成串行，以此来控制同一时间内共享数据只能被一个任务所修改，进而保证数据安全。

可以肯定的一点是：保护不同的数据的安全，就应该加不同的锁。

要想了解GIL，首先确定一点：每次执行python程序，都会产生一个独立的进程。例如python test.py，python aaa.py，python bbb.py会产生3个不同的python进程

#test.py内容
import os,time
print(os.getpid())
time.sleep(1000)

#打开终端执行
python3 test.py

#在windows下查看
tasklist |findstr python

#在linux下下查看
ps aux |grep python

在一个python的进程内，不仅有test.py的主线程或者由该主线程开启的其他线程，还有解释器开启的垃圾回收等解释器级别的线程，总之，所有线程都运行在这一个进程内，毫无疑问

1. 所有数据都是共享的，这其中，代码作为一种数据也是被所有线程共享的（test.py的所有代码以及Cpython解释器的所有代码）
   例如：test.py定义一个函数work（代码内容如下图），在进程内所有线程都能访问到work的代码，于是我们可以开启三个线程然后target都指向该代码，能访问到意味着就是可以执行。
2. 所有线程的任务，都需要将任务的代码当做参数传给解释器的代码去执行，即所有的线程要想运行自己的任务，首先需要解决的是能够访问到解释器的代码。

综上：
如果多个线程的target=work，那么执行流程是:
多个线程先访问到解释器的代码，即拿到执行权限，然后将target的代码交给解释器的代码去执行

解释器的代码是所有线程共享的，所以垃圾回收线程也可能访问到解释器的代码而去执行，这就导致了一个问题:对于同一个数据100，可能线程1执行x=100的同时，而垃圾回收执行的是回收100的操作，解决这种问题没有什么高明的方法，就是加锁处理，如下图的GIL，保证python解释器同一时间只能执行一个任务的代码

GIL.png

GIL和LOCK
机智的同学可能会问到这个问题：Python已经有一个GIL来保证同一时间只能有一个线程来执行了，为什么这里还需要lock?
首先，我们需要达成共识：锁的目的是为了保护共享的数据，同一时间只能有一个线程来修改共享的数据
然后，我们可以得出结论：保护不同的数据就应该加不同的锁。
最后，问题就很明朗了，GIL 与Lock是两把锁，保护的数据不一样，前者是解释器级别的（当然保护的就是解释器级别的数据，比如垃圾回收的数据），后者是保护用户自己开发的应用程序的数据，很明显GIL不负责这件事，只能用户自定义加锁处理，即Lock，如下图：

protect-data.png

分析：

1. 100个线程去抢GIL锁，即抢执行权限
2. 肯定有一个线程先抢到GIL（暂且称为线程1），然后开始执行，一旦执行就会拿到lock.acquire()
3. 极有可能线程1还未运行完毕，就有另外一个线程2抢到GIL，然后开始运行，但线程2发现互斥锁lock还未被线程1释放，于是阻塞，被迫交出执行权限，即释放GIL
4. 直到线程1重新抢到GIL，开始从上次暂停的位置继续执行，直到正常释放互斥锁lock，然后其他的线程再重复2 3 4的过程

from threading import Thread,Lock
import os,time
def work():
    global n
    lock.acquire()
    temp=n
    time.sleep(0.1)
    n=temp-1
    lock.release()
if __name__ == '__main__':
    lock=Lock()
    n=100
    l=[]
    for i in range(100):
        p=Thread(target=work)
        l.append(p)
        p.start()
    for p in l:
        p.join()

    print(n) #结果肯定为0，由原来的并发执行变成串行，牺牲了执行效率保证了数据安全，不加锁则结果可能为99

多线程用于IO密集型，如socket，爬虫，web
多进程用于计算密集型，如金融分析

死锁和递归锁

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

from threading import Thread,Lock
import time
mutexA=Lock()
mutexB=Lock()

class MyThread(Thread):
    def run(self):
        self.func1()
        self.func2()
    def func1(self):
        mutexA.acquire()
        print('\033[41m%s 拿到A锁\033[0m' %self.name)

        mutexB.acquire()
        print('\033[42m%s 拿到B锁\033[0m' %self.name)
        mutexB.release()

        mutexA.release()

    def func2(self):
        mutexB.acquire()
        print('\033[43m%s 拿到B锁\033[0m' %self.name)
        time.sleep(2)

        mutexA.acquire()
        print('\033[44m%s 拿到A锁\033[0m' %self.name)
        mutexA.release()

        mutexB.release()

if __name__ == '__main__':
    for i in range(10):
        t=MyThread()
        t.start()

递归锁
解决方法，递归锁，在Python中为了支持在同一线程中多次请求同一资源，python提供了可重入锁RLock。

这个RLock内部维护着一个Lock和一个counter变量，counter记录了acquire的次数，从而使得资源可以被多次require。直到一个线程所有的acquire都被release，其他的线程才能获得资源。上面的例子如果使用RLock代替Lock，则不会发生死锁，二者的区别是：递归锁可以连续acquire多次，而互斥锁只能acquire一次

from threading import Thread,RLock
import time

mutexA=mutexB=RLock() #一个线程拿到锁，counter加1,该线程内又碰到加锁的情况，则counter继续加1，这期间所有其他线程都只能等待，等待该线程释放所有锁，即counter递减到0为止

class MyThread(Thread):
    def run(self):
        self.func1()
        self.func2()
    def func1(self):
        mutexA.acquire()
        print('\033[41m%s 拿到A锁\033[0m' %self.name)

        mutexB.acquire()
        print('\033[42m%s 拿到B锁\033[0m' %self.name)
        mutexB.release()

        mutexA.release()

    def func2(self):
        mutexB.acquire()
        print('\033[43m%s 拿到B锁\033[0m' %self.name)
        time.sleep(2)

        mutexA.acquire()
        print('\033[44m%s 拿到A锁\033[0m' %self.name)
        mutexA.release()

        mutexB.release()

if __name__ == '__main__':
    for i in range(10):
        t=MyThread()
        t.start()

信号量，Event，定时器

信号量
信号量也是一把锁，可以指定信号量为5，对比互斥锁同一时间只能有一个任务抢到锁去执行，信号量同一时间可以有5个任务拿到锁去执行，如果说互斥锁是合租房屋的人去抢一个厕所，那么信号量就相当于一群路人争抢公共厕所，公共厕所有多个坑位，这意味着同一时间可以有多个人上公共厕所，但公共厕所容纳的人数是一定的，这便是信号量的大小

from threading import Thread,Semaphore
import threading
import time

def func():
    sm.acquire()
    print('%s get sm' %threading.current_thread().getName())
    time.sleep(3)
    sm.release()

if __name__ == '__main__':
    sm=Semaphore(5)
    for i in range(23):
        t=Thread(target=func)
        t.start()

'''
Semaphore管理一个内置的计数器，
每当调用acquire()时内置计数器-1；
调用release() 时内置计数器+1；
计数器不能小于0；当计数器为0时，acquire()将阻塞线程直到其他线程调用release()。
'''

Event
线程的一个关键特性是每个线程都是独立运行且状态不可预测。如果程序中的其 他线程需要通过判断某个线程的状态来确定自己下一步的操作,这时线程同步问题就会变得非常棘手。为了解决这些问题,我们需要使用threading库中的Event对象。 对象包含一个可由线程设置的信号标志,它允许线程等待某些事件的发生。在 初始情况下,Event对象中的信号标志被设置为假。如果有线程等待一个Event对象, 而这个Event对象的标志为假,那么这个线程将会被一直阻塞直至该标志为真。一个线程如果将一个Event对象的信号标志设置为真,它将唤醒所有等待这个Event对象的线程。如果一个线程等待一个已经被设置为真的Event对象,那么它将忽略这个事件, 继续执行

from threading import Event

event.isSet()：返回event的状态值；

event.wait()：如果 event.isSet()==False将阻塞线程；

event.set()： 设置event的状态值为True，所有阻塞池的线程激活进入就绪状态， 等待操作系统调度；

event.clear()：恢复event的状态值为False。

例如，有多个工作线程尝试链接MySQL，我们想要在链接前确保MySQL服务正常才让那些工作线程去连接MySQL服务器，如果连接不成功，都会去尝试重新连接。那么我们就可以采用threading.Event机制来协调各个工作线程的连接操作

from threading import Thread,Event
import threading
import time,random
def conn_mysql():
    count=1
    while not event.is_set():
        if count > 3:
            raise TimeoutError('链接超时')
        print('<%s>第%s次尝试链接' % (threading.current_thread().getName(), count))
        event.wait(0.5)
        count+=1
    print('<%s>链接成功' %threading.current_thread().getName())


def check_mysql():
    print('\033[45m[%s]正在检查mysql\033[0m' % threading.current_thread().getName())
    time.sleep(random.randint(2,4))
    event.set()
if __name__ == '__main__':
    event=Event()
    conn1=Thread(target=conn_mysql)
    conn2=Thread(target=conn_mysql)
    check=Thread(target=check_mysql)

    conn1.start()
    conn2.start()
    check.start()

定时器

from threading import Timer

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

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

线程队列

线程queue
queue is especially useful in threaded programming when information must be exchanged safely between multiple threads.
有三种不同的用法

• class queue.Queue(maxsize=0) 先进先出

import queue

q=queue.Queue()
q.put('first')
q.put('second')
q.put('third')

print(q.get())
print(q.get())
print(q.get())



'''
结果(先进先出):
first
second
third
'''

• class queue.LifoQueue(maxsize=0) last in first out

import queue

q=queue.LifoQueue()
q.put('first')
q.put('second')
q.put('third')

print(q.get())
print(q.get())
print(q.get())



'''
结果(后进先出):
third
second
first
'''

• class queue.PriorityQueue(maxsize=0) 优先队列：存储时可以设置优先级队列

import queue

q=queue.PriorityQueue()
#put进入一个元组,元组的第一个元素是优先级(通常是数字,也可以是非数字之间的比较),数字越小优先级越高
q.put((20,'a'))
q.put((10,'b'))
q.put((30,'c'))

print(q.get())
print(q.get())
print(q.get())



'''
结果(数字越小优先级越高,优先级高的优先出队):
(10, 'b')
(20, 'a')
(30, 'c')
'''

进程池与线城池

在刚开始学多进程或多线程时，我们迫不及待地基于多进程或多线程实现并发的套接字通信，然而这种实现方式的致命缺陷是：服务的开启的进程数或线程数都会随着并发的客户端数目地增多而增多，这会对服务端主机带来巨大的压力，甚至于不堪重负而瘫痪，于是我们必须对服务端开启的进程数或线程数加以控制，让机器在一个自己可以承受的范围内运行，这就是进程池或线程池的用途，例如进程池，就是用来存放进程的池子，本质还是基于多进程，只不过是对开启进程的数目加上了限制

介绍：
官网：https://docs.python.org/dev/library/concurrent.futures.html
concurrent.futures模块提供了高度封装的异步调用接口
ThreadPoolExecutor：线程池，提供异步调用
ProcessPoolExecutor: 进程池，提供异步调用
Both implement the same interface, which is defined by the abstract Executor class.

基本方法：

1、submit(fn, *args, **kwargs)
异步提交任务

2、map(func, *iterables, timeout=None, chunksize=1) 
取代for循环submit的操作

3、shutdown(wait=True) 
相当于进程池的pool.close()+pool.join()操作
wait=True，等待池内所有任务执行完毕回收完资源后才继续
wait=False，立即返回，并不会等待池内的任务执行完毕
但不管wait参数为何值，整个程序都会等到所有任务执行完毕
submit和map必须在shutdown之前

4、result(timeout=None)
取得结果

5、add_done_callback(fn)
回调函数

进程池
介绍：
The ProcessPoolExecutor class is an Executor subclass that uses a pool of processes to execute calls asynchronously. ProcessPoolExecutor uses the multiprocessing module, which allows it to side-step the Global Interpreter Lock but also means that only picklable objects can be executed and returned.

class concurrent.futures.ProcessPoolExecutor(max_workers=None, mp_context=None)
An Executor subclass that executes calls asynchronously using a pool of at most max_workers processes. If max_workers is None or not given, it will default to the number of processors on the machine. If max_workers is lower or equal to 0, then a ValueError will be raised.

用法：

from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor

import os,time,random
def task(n):
    print('%s is runing' %os.getpid())
    time.sleep(random.randint(1,3))
    return n**2

if __name__ == '__main__':

    executor=ProcessPoolExecutor(max_workers=3)

    futures=[]
    for i in range(11):
        future=executor.submit(task,i)
        futures.append(future)
    executor.shutdown(True)
    print('+++>')
    for future in futures:
        print(future.result())

线程池
ThreadPoolExecutor is an Executor subclass that uses a pool of threads to execute calls asynchronously.
class concurrent.futures.ThreadPoolExecutor(max_workers=None, thread_name_prefix='')
An Executor subclass that uses a pool of at most max_workers threads to execute calls asynchronously.

Changed in version 3.5: If max_workers is None or not given, it will default to the number of processors on the machine, multiplied by 5, assuming that ThreadPoolExecutor is often used to overlap I/O instead of CPU work and the number of workers should be higher than the number of workers for ProcessPoolExecutor.

New in version 3.6: The thread_name_prefix argument was added to allow users to control the threading.Thread names for worker threads created by the pool for easier debugging.

用法：
把ProcessPoolExecutor换成ThreadPoolExecutor，其余用法全部相同

map方法

from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor

import os,time,random
def task(n):
    print('%s is runing' %os.getpid())
    time.sleep(random.randint(1,3))
    return n**2

if __name__ == '__main__':

    executor=ThreadPoolExecutor(max_workers=3)

    # for i in range(11):
    #     future=executor.submit(task,i)

    executor.map(task,range(1,12)) #map取代了for+submit

回调函数
可以为进程池或线程池内的每个进程或线程绑定一个函数，该函数在进程或线程的任务执行完毕后自动触发，并接收任务的返回值当作参数，该函数称为回调函数

from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
from multiprocessing import Pool
import requests
import json
import os

def get_page(url):
    print('<进程%s> get %s' %(os.getpid(),url))
    respone=requests.get(url)
    if respone.status_code == 200:
        return {'url':url,'text':respone.text}

def parse_page(res):
    res=res.result()
    print('<进程%s> parse %s' %(os.getpid(),res['url']))
    parse_res='url:<%s> size:[%s]\n' %(res['url'],len(res['text']))
    with open('db.txt','a') as f:
        f.write(parse_res)


if __name__ == '__main__':
    urls=[
        'https://www.baidu.com',
        'https://www.python.org',
        'https://www.openstack.org',
        'https://help.github.com/',
        'http://www.sina.com.cn/'
    ]

    p=ProcessPoolExecutor(3)
    for url in urls:
        p.submit(get_page,url).add_done_callback(parse_page) #parse_page拿到的是一个future对象obj，需要用obj.result()拿到结果

并发编程之协程

协程介绍

本节的主题是基于单线程来实现并发，即只用一个主线程（很明显可利用的cpu只有一个）情况下实现并发，为此我们需要先回顾下并发的本质：切换+保存状态

cpu正在运行一个任务，会在两种情况下切走去执行其他的任务（切换由操作系统强制控制），一种情况是该任务发生了阻塞，另外一种情况是该任务计算的时间过长或有一个优先级更高的程序替代了它

三种执行状态.png

单纯地切换反而会降低运行效率:

#!/usr/bin/env python3
# Author    : fbo
# @Time : 18-5-23 上午5:59
# @File : 10 协程.py

# 串行执行
import time
def consumer(res):
    '''
    任务１：接收数据,处理数据
    :param res:
    :return:
    '''
    pass

def producer():
    '''任务２：生产数据'''
    res = []
    for i in range(10000000):
        res.append(i)
    return res

start = time.time()
# 串行执行
res = producer()
consumer(res) # 写成consumer(producer())会降低执行那个效率
stop = time.time()
print(stop - start)

# 基于yield并发执行
import time

def consumer():
    while True:
        x = yield
        print(x)

def producer():
    g = consumer()
    next(g)
    for i in range(100000):
        g.send(i)

start = time.time()
# 基于yield保存状态,实现两个任务直接来回切换,即并发的效果
# PS:如果每个任务中都加上打印,那么明显地看到两个任务的打印
# 是你一次我一次,即并发执行的.
producer()
stop = time.time()
print(stop-start)

第一种情况的切换。在任务一遇到io情况下，切到任务二去执行，这样就可以利用任务一阻塞的时间完成任务二的计算，效率的提升就在于此。

yield并不能实现遇到io切换

import time
def consumer():
    '''任务1:接收数据,处理数据'''
    while True:
        x=yield

def producer():
    '''任务2:生产数据'''
    g=consumer()
    next(g)
    for i in range(10000000):
        g.send(i)
        time.sleep(2)

start=time.time()
producer() #并发执行,但是任务producer遇到io就会阻塞住,并不会切到该线程内的其他任务去执行

stop=time.time()
print(stop-start)

对于单线程下，我们不可避免程序中出现io操作，但如果我们能在自己的程序中（即用户程序级别，而非操作系统级别）控制单线程下的多个任务能在一个任务遇到io阻塞时就切换到另外一个任务去计算，这样就保证了该线程能够最大限度地处于就绪态，即随时都可以被cpu执行的状态，相当于我们在用户程序级别将自己的io操作最大限度地隐藏起来，从而可以迷惑操作系统，让其看到：该线程好像是一直在计算，io比较少，从而更多的将cpu的执行权限分配给我们的线程。

协程的本质就是在单线程下，由用户自己控制一个任务遇到io阻塞了就切换另外一个任务去执行，以此来提升效率。为了实现它，我们需要找寻一种可以同时满足以下条件的解决方案：

1. 可以控制多个任务之间的切换，切换之前将任务的状态保存下来，以便重新运行时，可以基于暂停的位置继续执行。
2. 作为1的补充：可以检测io操作，在遇到io操作的情况下才发生切换

协程介绍
协程：是单线程下的并发，又称微线程，纤程。英文名Coroutine。一句话说明什么是线程：协程是一种用户态的轻量级线程，即协程是由用户程序自己控制调度的。

需要强调的是：

1. python的线程属于内核级别的，即由操作系统控制调度（如单线程遇到io或执行时间过长就会被迫交出cpu执行权限，切换其他线程运行）
2. 单线程内开启协程，一旦遇到io，就会从应用程序级别（而非操作系统）控制切换，以此来提升效率（！！！非io操作的切换与效率无关）

对比操作系统控制线程的切换，用户在单线程内控制协程的切换，优点如下：

1. 协程的切换开销更小，属于程序级别的切换，操作系统完全感知不到，因而更加轻量级
2. 单线程内就可以实现并发的效果，最大限度地利用cpu

缺点如下：

1. 协程的本质是单线程下，无法利用多核，可以是一个程序开启多个进程，每个进程内开启多个线程，每个线程内开启协程
2. 协程指的是单个线程，因而一旦协程出现阻塞，将会阻塞整个线程

总结协程的特点：

1. 必须只有一个单线程里实现并发
2. 修改共享数据不需要加锁
3. 用户程序里自己保存多个控制流的上下文栈
4. 一个协程里遇到IO操作自动切换到其他协程（如何实现检测IO， yield、greenlet都无法实现， 就用到了gevent模块（select机制））

greenlet模块

如果我们在单个线程内有20个任务，要想实现在多个任务之间切换，使用yield生成器的方式过于麻烦（需要先得到初始化一次的生成器，然后再调用send。。。非常麻烦），而使用greenlet模块可以非常简单地实现这20个任务直接的切换

#安装：pip3 install greenlet
from greenlet import greenlet

def eat(name):
    print('%s eat 1' %name)
    g2.switch('egon')
    print('%s eat 2' %name)
    g2.switch()
def play(name):
    print('%s play 1' %name)
    g1.switch()
    print('%s play 2' %name)

g1=greenlet(eat)
g2=greenlet(play)

g1.switch('egon')#可以在第一次switch时传入参数，以后都不需要

单纯的切换（在没有io的情况下或者没有重复开辟内存空间的操作），反而会降低程序的执行速度

#顺序执行
import time
def f1():
    res=1
    for i in range(100000000):
        res+=i

def f2():
    res=1
    for i in range(100000000):
        res*=i

start=time.time()
f1()
f2()
stop=time.time()
print('run time is %s' %(stop-start)) #10.985628366470337

#切换
from greenlet import greenlet
import time
def f1():
    res=1
    for i in range(100000000):
        res+=i
        g2.switch()

def f2():
    res=1
    for i in range(100000000):
        res*=i
        g1.switch()

start=time.time()
g1=greenlet(f1)
g2=greenlet(f2)
g1.switch()
stop=time.time()
print('run time is %s' %(stop-start)) # 52.763017892837524

greenlet只是提供了一种比generator更加便捷的切换方式，当切到一个任务执行时如果遇到io，那就原地阻塞，仍然是没有解决遇到IO自动切换来提升效率的问题。

单线程里的这20个任务的代码通常会既有计算操作又有阻塞操作，我们完全可以在执行任务1时遇到阻塞，就利用阻塞的时间去执行任务2。。。。如此，才能提高效率，这就用到了Gevent模块。

gevent模块

Gevent 是一个第三方库，可以轻松通过gevent实现并发同步或异步编程，在gevent中用到的主要模式是Greenlet, 它是以C扩展模块形式接入Python的轻量级协程。 Greenlet全部运行在主程序操作系统进程的内部，但它们被协作式地调度。

用法：

g1=gevent.spawn(func,1,,2,3,x=4,y=5)创建一个协程对象g1，spawn括号内第一个参数是函数名，如eat，后面可以有多个参数，可以是位置实参或关键字实参，都是传给函数eat的

g2=gevent.spawn(func2)

g1.join() #等待g1结束

g2.join() #等待g2结束
#或者上述两步合作一步：gevent.joinall([g1,g2])

g1.value#拿到func1的返回值

遇到IO阻塞时会自动切换任务

import gevent
def eat(name):
    print('%s eat 1' %name)
    gevent.sleep(2)
    print('%s eat 2' %name)

def play(name):
    print('%s play 1' %name)
    gevent.sleep(1)
    print('%s play 2' %name)


g1=gevent.spawn(eat,'egon')
g2=gevent.spawn(play,name='egon')
g1.join()
g2.join()
#或者gevent.joinall([g1,g2])
print('主')

上例gevent.sleep(2)模拟的是gevent可以识别的io阻塞,

而time.sleep(2)或其他的阻塞,gevent是不能直接识别的需要用下面一行代码,打补丁,就可以识别了

from gevent import monkey;monkey.patch_all()必须放到被打补丁者的前面，如time，socket模块之前

或者我们干脆记忆成：要用gevent，需要将from gevent import monkey;monkey.patch_all()放到文件的开头

练习
服务端：

from gevent import monkey;monkey.patch_all()
from socket import *
import gevent

#如果不想用money.patch_all()打补丁,可以用gevent自带的socket
# from gevent import socket
# s=socket.socket()

def server(server_ip,port):
    s=socket(AF_INET,SOCK_STREAM)
    s.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
    s.bind((server_ip,port))
    s.listen(5)
    while True:
        conn,addr=s.accept()
        gevent.spawn(talk,conn,addr)

def talk(conn,addr):
    try:
        while True:
            res=conn.recv(1024)
            print('client %s:%s msg: %s' %(addr[0],addr[1],res))
            conn.send(res.upper())
    except Exception as e:
        print(e)
    finally:
        conn.close()

if __name__ == '__main__':
    server('127.0.0.1',8080)

多线程并发多个客户端

from threading import Thread
from socket import *
import threading

def client(server_ip,port):
    c=socket(AF_INET,SOCK_STREAM) #套接字对象一定要加到函数内，即局部名称空间内，放在函数外则被所有线程共享，则大家公用一个套接字对象，那么客户端端口永远一样了
    c.connect((server_ip,port))

    count=0
    while True:
        c.send(('%s say hello %s' %(threading.current_thread().getName(),count)).encode('utf-8'))
        msg=c.recv(1024)
        print(msg.decode('utf-8'))
        count+=1
if __name__ == '__main__':
    for i in range(500):
        t=Thread(target=client,args=('127.0.0.1',8080))
        t.start()

IO模型

http://www.cnblogs.com/linhaifeng/articles/7430066.html#_label4

同步（synchronous） IO和异步（asynchronous） IO，阻塞（blocking） IO和非阻塞（non-blocking）IO分别是什么，到底有什么区别？这个问题其实不同的人给出的答案都可能不同，比如wiki，就认为asynchronous IO和non-blocking IO是一个东西。这其实是因为不同的人的知识背景不同，并且在讨论这个问题的时候上下文(context)也不相同。所以，为了更好的回答这个问题，我先限定一下本文的上下文。

本文讨论的背景是Linux环境下的network IO。本文最重要的参考文献是Richard Stevens的“UNIX® Network Programming Volume 1, Third Edition: The Sockets Networking ”，6.2节“I/O Models ”，Stevens在这节中详细说明了各种IO的特点和区别，如果英文够好的话，推荐直接阅读。Stevens的文风是有名的深入浅出，所以不用担心看不懂。本文中的流程图也是截取自参考文献。

Stevens在文章中一共比较了五种IO Model：

• blocking IO
• nonblocking IO
• IO multiplexing
• signal driven IO
• asynchronous IO
  由signal driven IO（信号驱动IO）在实际中并不常用，所以主要介绍其余四种IO Model。

再说一下IO发生时涉及的对象和步骤。对于一个network IO (这里我们以read举例)，它会涉及到两个系统对象，一个是调用这个IO的process (or thread)，另一个就是系统内核(kernel)。当一个read操作发生时，该操作会经历两个阶段：

1. 等待数据准备 (Waiting for the data to be ready)
2. 将数据从内核拷贝到进程中(Copying the data from the kernel to the process)

记住这两点很重要，因为这些IO模型的区别就是在两个阶段上各有不同的情况。

阻塞IO （blocking IO）

在linux中，默认情况下所有的socket都是blocking，一个典型的读操作流程大概是这样的：

blocking_io.png

当用户进程调用了recvfrom这个系统调用，kernel就开始了IO的第一个阶段：准备数据。对于network io来说，很多时候数据在一开始还未到达，这个时候kernel就要等待足够的数据到来；
而在用户进程这边，整个进程会被阻塞。当kernel一直等到数据准备好，就会将数据从kernel中拷贝到用户内存，然后kernel返回结果，用户进程才接错block的状态，重新运行起来。
所以，blocking IO的特点就是在IO执行的两个阶段（等待数据和拷贝数据的两个阶段）都被block了

几乎所有的程序员第一次接触到的网络编程都是从listen()、send()、recv() 等接口开始的，
使用这些接口可以很方便的构建服务器/客户机的模型。然而大部分的socket接口都是阻塞型的。如下图

image.png

ps：
所谓阻塞型接口是指系统调用（一般是IO接口）不返回调用结果并让当前线程一直阻塞
只有当该系统调用获得结果或者超时出错时才返回。

实际上，除非特别指定，几乎所有的IO接口 ( 包括socket接口 ) 都是阻塞型的。这给网络编程带来了一个很大的问题，如在调用recv(1024)的同时，线程将被阻塞，在此期间，线程将无法执行任何运算或响应任何的网络请求。

一个简单的解决方案：
在服务器端使用多线程（或多进程）。多线程（或多进程）的目的是让每个连接都拥有独立的线程（或进程），
这样任何一个连接的阻塞都不会影响其他的连接。

该方案的问题是：
开启多进程或都线程的方式，在遇到要同时响应成百上千路的连接请求，则无论多线程还是多进程都会严重占据系统资源，
降低系统对外界响应效率，而且线程与进程本身也更容易进入假死状态。

改进方案：
很多程序员可能会考虑使用“线程池”或“连接池”。“线程池”旨在减少创建和销毁线程的频率，
其维持一定合理数量的线程，并让空闲的线程重新承担新的执行任务。“连接池”维持连接的缓存池，尽量重用已有的连接、
减少创建和关闭连接的频率。这两种技术都可以很好的降低系统开销，都被广泛应用很多大型系统，如websphere、tomcat和各种数据库等。

改进后方案其实也存在着问题：
“线程池”和“连接池”技术也只是在一定程度上缓解了频繁调用IO接口带来的资源占用。而且，所谓“池”始终有其上限，
当请求大大超过上限时，“池”构成的系统对外界的响应并不比没有池的时候效果好多少。所以使用“池”必须考虑其面临的响应规模，
并根据响应规模调整“池”的大小。

对应上例中的所面临的可能同时出现的上千甚至上万次的客户端请求，“线程池”或“连接池”或许可以缓解部分压力，但是不能解决所有问题。总之，多线程模型可以方便高效的解决小规模的服务请求，但面对大规模的服务请求，多线程模型也会遇到瓶颈，可以用非阻塞接口来尝试解决这个问题。

非阻塞IO （non-blocking IO）

Linux下，可以通过设置socket使其变为non-blocking。当对一个non-blocking socket执行读操作时，流程是这个样子：

none-blocking.png

示例：

#服务端
from socket import *

server = socket(AF_INET, SOCK_STREAM)
server.bind(('127.0.0.1',8099))
server.listen(5)
server.setblocking(False)


rlist=[]
wlist=[]
while True:
    try:
        conn, addr = server.accept()
        rlist.append(conn)
        print(rlist)

    except BlockingIOError:
        del_rlist=[]
        for sock in rlist:
            try:
                data=sock.recv(1024)
                if not data:
                    del_rlist.append(sock)
                wlist.append((sock,data.upper()))
            except BlockingIOError:
                continue
            except Exception:
                sock.close()
                del_rlist.append(sock)

        del_wlist=[]
        for item in wlist:
            try:
                sock = item[0]
                data = item[1]
                sock.send(data)
                del_wlist.append(item)
            except BlockingIOError:
                pass

        for item in del_wlist:
            wlist.remove(item)


        for sock in del_rlist:
            rlist.remove(sock)

server.close()


#客户端
from socket import *
c=socket(AF_INET,SOCK_STREAM)
c.connect(('127.0.0.1',8080))

while True:
    msg=input('>>: ')
    if not msg:continue
    c.send(msg.encode('utf-8'))
    data=c.recv(1024)
    print(data.decode('utf-8'))

但是非阻塞IO模型绝不被推荐。

我们不能否则其优点：能够在等待任务完成的时间里干其他活了（包括提交其他任务，也就是 “后台” 可以有多个任务在“”同时“”执行）。
但是也难掩其缺点：

1. 循环调用recv()将大幅度推高CPU占用率；这也是我们在代码中留一句time.sleep(2)的原因,否则在低配主机下极容易出现卡机情况
2. 任务完成的响应延迟增大了，因为每过一段时间才去轮询一次read操作，而任务可能在两次轮询之间的任意时间完成。
   这会导致整体数据吞吐量的降低。

此外，在这个方案中recv()更多的是起到检测“操作是否完成”的作用，实际操作系统提供了更为高效的检测“操作是否完成“作用的接口，例如select()多路复用模式，可以一次检测多个连接是否活跃。

多路复用IO（IO multiplexing）

IO multiplexing这个词可能有点陌生，但是如果我说select/epoll，大概就都能明白了。有些地方也称这种IO方式为事件驱动IO（event driven IO）。我们都知道，select/epoll的好处就在于单个process就可以同时处理多个网络连接的IO。它的基本原理就是select/epoll这个function会不断的轮询所负责的所有socket，当某个socket有数据到达了，就通知用户进程。它的流程如图：

image.png

强调：

1. 如果处理的连接数不是很高的话，使用select/epoll的web server不一定比使用multi-threading + blocking IO的web server性能更好，可能延迟还更大。select/epoll的优势并不是对于单个连接能处理得更快，而是在于能处理更多的连接。
2. 在多路复用模型中，对于每一个socket，一般都设置成为non-blocking，但是，如上图所示，整个用户的process其实是一直被block的。只不过process是被select这个函数block，而不是被socket IO给block。

结论: select的优势在于可以处理多个连接，不适用于单个连接

示例：

#服务端
from socket import *
import select
server = socket(AF_INET, SOCK_STREAM)
server.bind(('127.0.0.1',8093))
server.listen(5)
server.setblocking(False)
print('starting...')

rlist=[server,]
wlist=[]
wdata={}

while True:
    rl,wl,xl=select.select(rlist,wlist,[],0.5)
    print(wl)
    for sock in rl:
        if sock == server:
            conn,addr=sock.accept()
            rlist.append(conn)
        else:
            try:
                data=sock.recv(1024)
                if not data:
                    sock.close()
                    rlist.remove(sock)
                    continue
                wlist.append(sock)
                wdata[sock]=data.upper()
            except Exception:
                sock.close()
                rlist.remove(sock)

    for sock in wl:
        sock.send(wdata[sock])
        wlist.remove(sock)
        wdata.pop(sock)


#客户端
from socket import *
c=socket(AF_INET,SOCK_STREAM)
c.connect(('127.0.0.1',8081))

while True:
    msg=input('>>: ')
    if not msg:continue
    c.send(msg.encode('utf-8'))
    data=c.recv(1024)
    print(data.decode('utf-8'))

select监听fd变化的过程分析：
用户进程创建socket对象，拷贝监听的fd到内核空间，每一个fd会对应一张系统文件表，内核空间的fd响应到数据后，就会发送信号给用户进程数据已到；
用户进程再发送系统调用，比如（accept）将内核空间的数据copy到用户空间，同时作为接受数据端内核空间的数据清除，这样重新监听时fd再有新的数据又可以响应到了（发送端因为基于TCP协议所以需要收到应答后才会清除）。

该模型的优点：
相比其他模型，使用select() 的事件驱动模型只用单线程（进程）执行，占用资源少，不消耗太多 CPU，同时能够为多客户端提供服务。
如果试图建立一个简单的事件驱动的服务器程序，这个模型有一定的参考价值。

该模型的缺点：
首先select()接口并不是实现“事件驱动”的最好选择。因为当需要探测的句柄值较大时，select()接口本身需要消耗大量时间去轮询各个句柄。
很多操作系统提供了更为高效的接口，如linux提供了epoll，BSD提供了kqueue，Solaris提供了/dev/poll，…。
如果需要实现更高效的服务器程序，类似epoll这样的接口更被推荐。遗憾的是不同的操作系统特供的epoll接口有很大差异，
所以使用类似于epoll的接口实现具有较好跨平台能力的服务器会比较困难。
其次，该模型将事件探测和事件响应夹杂在一起，一旦事件响应的执行体庞大，则对整个模型是灾难性的。

异步IO(Asynchronous I/O)

Linux下的asynchronous IO其实用得不多，从内核2.6版本才开始引入。先看一下它的流程：

image.png

用户进程发起read操作之后，立刻就可以开始去做其它的事。而另一方面，从kernel的角度，当它受到一个asynchronous read之后，首先它会立刻返回，所以不会对用户进程产生任何block。然后，kernel会等待数据准备完成，然后将数据拷贝到用户内存，当这一切都完成之后，kernel会给用户进程发送一个signal，告诉它read操作完成了。

IO模型分析比较

到目前为止，已经将四个IO Model都介绍完了。现在回过头来回答最初的那几个问题：blocking和non-blocking的区别在哪，synchronous IO和asynchronous IO的区别在哪。
先回答最简单的这个：blocking vs non-blocking。前面的介绍中其实已经很明确的说明了这两者的区别。调用blocking IO会一直block住对应的进程直到操作完成，而non-blocking IO在kernel还准备数据的情况下会立刻返回。

再说明synchronous IO和asynchronous IO的区别之前，需要先给出两者的定义。Stevens给出的定义（其实是POSIX的定义）是这样子的：
A synchronous I/O operation causes the requesting process to be blocked until that I/O operationcompletes;
An asynchronous I/O operation does not cause the requesting process to be blocked;
两者的区别就在于synchronous IO做”IO operation”的时候会将process阻塞。按照这个定义，四个IO模型可以分为两大类，
之前所述的blocking IO，non-blocking IO，IO multiplexing都属于synchronous IO这一类，而 asynchronous I/O后一类 。

有人可能会说，non-blocking IO并没有被block啊。这里有个非常“狡猾”的地方，定义中所指的”IO operation”是指真实的IO操作，
就是例子中的recvfrom这个system call。non-blocking IO在执行recvfrom这个system call的时候，如果kernel的数据没有准备好，
这时候不会block进程。但是，当kernel中数据准备好的时候，recvfrom会将数据从kernel拷贝到用户内存中，这个时候进程是被block了，
在这段时间内，进程是被block的。而asynchronous IO则不一样，当进程发起IO 操作之后，就直接返回再也不理睬了，直到kernel发送一个信号，
告诉进程说IO完成。在这整个过程中，进程完全没有被block。

各个IO Model的比较如图所示：

comparison.png

经过上面的介绍，会发现non-blocking IO和asynchronous IO的区别还是很明显的。在non-blocking IO中，虽然进程大部分时间都不会被block，但是它仍然要求进程去主动的check，并且当数据准备完成以后，也需要进程主动的再次调用recvfrom来将数据拷贝到用户内存。而asynchronous IO则完全不同。它就像是用户进程将整个IO操作交给了他人（kernel）完成，然后他人做完后发信号通知。在此期间，用户进程不需要去检查IO操作的状态，也不需要主动的去拷贝数据。

selectors模型

了解select，poll，epoll

IO复用：为了解释这个名词，首先来理解下复用这个概念，复用也就是共用的意思，这样理解还是有些抽象，为此，咱们来理解下复用在通信领域的使用，在通信领域中为了充分利用网络连接的物理介质，往往在同一条网络链路上采用时分复用或频分复用的技术使其在同一链路上传输多路信号，到这里我们就基本上理解了复用的含义，即公用某个“介质”来尽可能多的做同一类(性质)的事，那IO复用的“介质”是什么呢？为此我们首先来看看服务器编程的模型，客户端发来的请求服务端会产生一个进程来对其进行服务，每当来一个客户请求就产生一个进程来服务，然而进程不可能无限制的产生，因此为了解决大量客户端访问的问题，引入了IO复用技术，即：一个进程可以同时对多个客户请求进行服务。

也就是说IO复用的“介质”是进程(准确的说复用的是select和poll，因为进程也是靠调用select和poll来实现的)，

复用一个进程(select和poll)来对多个IO进行服务，虽然客户端发来的IO是并发的但是IO所需的读写数据多数情况下是没有准备好的，
因此就可以利用一个函数(select和poll)来监听IO所需的这些数据的状态，一旦IO有数据可以进行读写了，进程就来对这样的IO进行服务。

理解完IO复用后，我们在来看下实现IO复用中的三个API(select、poll和epoll)的区别和联系

select，poll，epoll都是IO多路复用的机制，I/O多路复用就是通过一种机制，可以监视多个描述符，一旦某个描述符就绪（一般是读就绪或者写就绪），能够通知应用程序进行相应的读写操作。但select，poll，epoll本质上都是同步I/O，因为他们都需要在读写事件就绪后自己负责进行读写，也就是说这个读写过程是阻塞的，而异步I/O则无需自己负责进行读写，异步I/O的实现会负责把数据从内核拷贝到用户空间。三者的原型如下所示：

int select(int nfds, fd_set *readfds, fd_set *writefds, fd_set *exceptfds, struct timeval *timeout);

int poll(struct pollfd *fds, nfds_t nfds, int timeout);

int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout);

1. select的第一个参数nfds为fdset集合中最大描述符值加1，fdset是一个位数组，其大小限制为__FD_SETSIZE（1024），位数组的每一位代表其对应的描述符是否需要被检查。第二三四参数表示需要关注读、写、错误事件的文件描述符位数组，这些参数既是输入参数也是输出参数，可能会被内核修改用于标示哪些描述符上发生了关注的事件，所以每次调用select前都需要重新初始化fdset。timeout参数为超时时间，该结构会被内核修改，其值为超时剩余的时间。
   select的调用步骤如下：
   （1）使用copy_from_user从用户空间拷贝fdset到内核空间
   （2）注册回调函数__pollwait
   （3）遍历所有fd，调用其对应的poll方法（对于socket，这个poll方法是sock_poll，sock_poll根据情况会调用到tcp_poll,udp_poll
   或者datagram_poll）
   （4）以tcp_poll为例，其核心实现就是__pollwait，也就是上面注册的回调函数。
   （5）__pollwait的主要工作就是把current（当前进程）挂到设备的等待队列中，不同的设备有不同的等待队列，对于tcp_poll 来说，其等待队列是sk->sk_sleep（注意把进程挂到等待队列中并不代表进程已经睡眠了）。在设备收到一条消息（网络设备）或填写完文件数据（磁盘设备）后，会唤醒设备等待队列上睡眠的进程，这时current便被唤醒了。
   （6）poll方法返回时会返回一个描述读写操作是否就绪的mask掩码，根据这个mask掩码给fd_set赋值。
   （7）如果遍历完所有的fd，还没有返回一个可读写的mask掩码，则会调用schedule_timeout是调用select的进程（也就是 current）进入睡眠。当设备驱动发生自身资源可读写后，会唤醒其等待队列上睡眠的进程。如果超过一定的超时时间（schedule_timeout 指定），还是没人唤醒，则调用select的进程会重新被唤醒获得CPU，进而重新遍历fd，判断有没有就绪的fd。
   （8）把fd_set从内核空间拷贝到用户空间。

总结下select的几大缺点：
（1）每次调用select，都需要把fd集合从用户态拷贝到内核态，这个开销在fd很多时会很大
（2）同时每次调用select都需要在内核遍历传递进来的所有fd，这个开销在fd很多时也很大
（3）select支持的文件描述符数量太小了，默认是1024

2． poll与select不同，通过一个pollfd数组向内核传递需要关注的事件，故没有描述符个数的限制，pollfd中的events字段和revents分别
用于标示关注的事件和发生的事件，故pollfd数组只需要被初始化一次。

poll的实现机制与select类似，其对应内核中的sys_poll，只不过poll向内核传递pollfd数组，然后对pollfd中的每个描述符进行poll，相比处理fdset来说，poll效率更高。poll返回后，需要对pollfd中的每个元素检查其revents值，来得指事件是否发生。

3．直到Linux2.6才出现了由内核直接支持的实现方法，那就是epoll，被公认为Linux2.6下性能最好的多路I/O就绪通知方法。
epoll可以同时支持水平触发和边缘触发（Edge Triggered，只告诉进程哪些文件描述符刚刚变为就绪状态，它只说一遍，如果我们没有采取行动，那么它将不会再次告知，这种方式称为边缘触发），理论上边缘触发的性能要更高一些，但是代码实现相当复杂。
epoll同样只告知那些就绪的文件描述符，而且当我们调用epoll_wait()获得就绪文件描述符时，返回的不是实际的描述符，而是一个代表就绪描述符数量的值，你只需要去epoll指定的一个数组中依次取得相应数量的文件描述符即可，这里也使用了内存映射（mmap）技术，这样便彻底省掉了这些文件描述符在系统调用时复制的开销。另一个本质的改进在于epoll采用基于事件的就绪通知方式。在select/poll中，进程只有在调用一定的方法后，内核才对所有监视的文件描述符进行扫描，而epoll事先通过epoll_ctl()来注册一个文件描述符，一旦基于某个文件描述符就绪时，内核会采用类似callback的回调机制，迅速激活这个文件描述符，当进程调用epoll_wait()时便得到通知。

epoll既然是对select和poll的改进，就应该能避免上述的三个缺点。那epoll都是怎么解决的呢？在此之前，我们先看一下epoll 和select和
poll的调用接口上的不同，select和poll都只提供了一个函数——select或者poll函数。而epoll提供了三个函 数，epoll_create,epoll_ctl和epoll_wait，epoll_create是创建一个epoll句柄；epoll_ctl是注 册要监听的事件类型；
epoll_wait则是等待事件的产生。

对于第一个缺点，epoll的解决方案在epoll_ctl函数中。每次注册新的事件到epoll句柄中时（在epoll_ctl中指定 EPOLL_CTL_ADD），会把所有的fd拷贝进内核，而不是在epoll_wait的时候重复拷贝。epoll保证了每个fd在整个过程中只会拷贝 一次。

对于第二个缺点，epoll的解决方案不像select或poll一样每次都把current轮流加入fd对应的设备等待队列中，而只在 epoll_ctl时把current挂一遍（这一遍必不可少）并为每个fd指定一个回调函数，当设备就绪，唤醒等待队列上的等待者时，就会调用这个回调 函数，而这个回调函数会把就绪的fd加入一个就绪链表）。epoll_wait的工作实际上就是在这个就绪链表中查看有没有就绪的fd（利用schedule_timeout()实现睡一会，判断一会的效果，和select实现中的第7步是类似的）。

对于第三个缺点，epoll没有这个限制，它所支持的FD上限是最大可以打开文件的数目，这个数字一般远大于2048,举个例子, 在1GB内存的机器上大约是10万左右，具体数目可以cat /proc/sys/fs/file-max察看,一般来说这个数目和系统内存关系很大。

总结：

1. select，poll实现需要自己不断轮询所有fd集合，直到设备就绪，期间可能要睡眠和唤醒多次交替。而epoll其实也需要调用 epoll_wait
   不断轮询就绪链表，期间也可能多次睡眠和唤醒交替，但是它是设备就绪时，调用回调函数，把就绪fd放入就绪链表中，并唤醒在 epoll_wait中进入睡眠的进程。虽然都要睡眠和交替，但是select和poll在“醒着”的时候要遍历整个fd集合，而epoll在“醒着”的 时候只要判断一下就绪链表是否为空就行了，这节省了大量的CPU时间，这就是回调机制带来的性能提升。
2. select，poll每次调用都要把fd集合从用户态往内核态拷贝一次，并且要把current往设备等待队列中挂一次，而epoll只要 一次拷贝，
   而且把current往等待队列上挂也只挂一次（在epoll_wait的开始，注意这里的等待队列并不是设备等待队列，只是一个epoll内
   部定义的等待队列），这也能节省不少的开销。

selectors模块
这三种IO多路复用模型在不同的平台有着不同的支持，而epoll在windows下就不支持，好在我们有selectors模块，帮我们默认选择当前平台下最合适的

#服务端
from socket import *
import selectors

sel=selectors.DefaultSelector()
def accept(server_fileobj,mask):
    conn,addr=server_fileobj.accept()
    sel.register(conn,selectors.EVENT_READ,read)

def read(conn,mask):
    try:
        data=conn.recv(1024)
        if not data:
            print('closing',conn)
            sel.unregister(conn)
            conn.close()
            return
        conn.send(data.upper()+b'_SB')
    except Exception:
        print('closing', conn)
        sel.unregister(conn)
        conn.close()



server_fileobj=socket(AF_INET,SOCK_STREAM)
server_fileobj.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
server_fileobj.bind(('127.0.0.1',8088))
server_fileobj.listen(5)
server_fileobj.setblocking(False) #设置socket的接口为非阻塞
sel.register(server_fileobj,selectors.EVENT_READ,accept) #相当于网select的读列表里append了一个文件句柄
                                                         #server_fileobj,并且绑定了一个回调函数accept

while True:
    events=sel.select() #检测所有的fileobj，是否有完成wait data的
    for sel_obj,mask in events:
        callback=sel_obj.data #callback=accpet
        callback(sel_obj.fileobj,mask) #accpet(server_fileobj,1)

#客户端
from socket import *
c=socket(AF_INET,SOCK_STREAM)
c.connect(('127.0.0.1',8088))

while True:
    msg=input('>>: ')
    if not msg:continue
    c.send(msg.encode('utf-8'))
    data=c.recv(1024)
    print(data.decode('utf-8'))

基于selectors模块实现并发的FTP

参考:链接: https://pan.baidu.com/s/1qYPrHCg 密码: 9is4

练习题

1、简述计算机操作系统中的“中断”的作用？

中断:计算机执行期间,系统内发生任何非寻常的或非预期的急需处理事件,使得cpu暂时中断当前正在执行的程序而转去执行相应的事件处理程序.
待处理完毕后又返回原来被中断处继续执行或调度新的进程执行的过程.
它使计算机可以更好更快利用有限的系统资源解决系统响应速度和运行效率的一种控制技术.
实时响应 + 系统调用

2、简述计算机内存中的“内核态”和“用户态”；

操作系统由操作系统的内核（运行于内核态，管理硬件资源）以及系统调用（运行于用户态，为应用程序员写的应用程序提供系统调用接口）两部分组成；

内核态：cpu可以访问内存的所有数据，包括外围设备，例如硬盘，网卡，cpu也可以将自己从一个程序切换到另一个程序。

用户态：只能受限的访问内存，且不允许访问外围设备，占用cpu的能力被剥夺，cpu资源可以被其他程序获取。

用户态的应用程序可以通过三种方式来访问内核态的资源：
1）系统调用
2）库函数
3）Shell脚本
用户态到内核态的切换:
1.系统调用 用户程序主动发起的 软中断 os.fork() process
2.异常 被动的 当CPU正在执行运行在用户态的程序时，突然发生某些预先不可知的异常事件，这个时候就会触发从当前用户态执行的进程
转向内核态执行相关的异常事件，典型的如缺页异常。
3.外围设备的硬中断 被动的 外围设备完成用户的请求操作后，会像CPU发出中断信号，此时，CPU就会暂停执行下一条即将要执行的指令，
转而去执行中断信号对应的处理程序，如果先前执行的指令是在用户态下，则自然就发生从用户态到内核态的转换。

image

详见：用户态和内核态的区别 https://blog.csdn.net/youngyoungla/article/details/53106671

https://blog.csdn.net/qq_34228570/article/details/72995997

3、进程间通信方式有哪些？

1、管道
2、有名管道（FIFO）
3、消息队列
4、信号量
5、共享内存
6、套接字（socket）

4、简述你对管道、队列的理解；

管道通常指无名管道
　　　　1、它是半双工的（即数据只能在一个方向上流动），具有固定的读端和写端
　　　　2、它只能用于具有亲缘关系的进程中通信（也就是父与子进程或者兄弟进程之间）
　　　　3、数据不可反复读取了，即读了之后欢喜红区中就没有了
　　消息队列
　　　　1、消息队列是面向记录的，其中的消息具有特定的格式以及特定的优先级
　　　　2、消息队列独立于发送与接收进程。进程终止时，消息队列及其内容不会被删除。
　　　　3、消息队列可以实现消息随机查询。

mutiprocessing模块为我们提供的基于消息的IPC通信机制：队列和管道。
　　队列和管道都是将数据存放于内存中，而队列又是基于（管道+锁）实现的，
可以让我们从复杂的锁问题中解脱出来，因而队列才是进程间通信的最佳选择。
　　我们应该尽量避免使用共享数据，尽可能使用消息传递和队列，避免处理复杂的同步和锁问题，
而且在进程数目增多时，往往可以获得更好的可展性。
队列 = 管道 + 锁

from multiprocessing import Queue,Process
queue = Queue()
queue.put(url)
url = queue.get()
from multiprocessing import Pipe,Process
pipe = Pipe()
pipe.send(url)
pipe.recv()

5、请列举你知道的进程间通信方式；

进程间通信(IPC)

1)管道

管道分为有名管道和无名管道

无名管道是一种半双工的通信方式,数据只能单向流动,而且只能在具有亲缘关系的进程间使用.进程的亲缘关系一般指的是父子关系。无明管道一般用于两个不同进程之间的通信。当一个进程创建了一个管道,并调用fork创建自己的一个子进程后,父进程关闭读管道端,子进程关闭写管道端,这样提供了两个进程之间数据流动的一种方式。

有名管道也是一种半双工的通信方式,但是它允许无亲缘关系进程间的通信。

2)信号量

信号量是一个计数器,可以用来控制多个线程对共享资源的访问.,它不是用于交换大批数据,而用于多线程之间的同步.它常作为一种锁机制,防止某进程在访问资源时其它进程也访问该资源.因此,主要作为进程间以及同一个进程内不同线程之间的同步手段.

3)信号

信号是一种比较复杂的通信方式,用于通知接收进程某个事件已经发生.

4)消息队列

消息队列是消息的链表,存放在内核中并由消息队列标识符标识.消息队列克服了信号传递信息少,管道只能承载无格式字节流以及缓冲区大小受限等特点.消息队列是UNIX下不同进程之间可实现共享资源的一种机制,UNIX允许不同进程将格式化的数据流以消息队列形式发送给任意进程.对消息队列具有操作权限的进程都可以使用msget完成对消息队列的操作控制.通过使用消息类型,进程可以按任何顺序读信息,或为消息安排优先级顺序.

5)共享内存

共享内存就是映射一段能被其他进程所访问的内存,这段共享内存由一个进程创建,但多个进程都可以访问.共享内存是最快的IPC(进程间通信)方式,它是针对其它进程间通信方式运行效率低而专门设计的.它往往与其他通信机制,如信号量,配合使用,来实现进程间的同步与通信.

6)套接字：可用于不同及其间的进程通信

6、什么是同步I/O，什么是异步I/O？

同步是指：发送方发出数据后，等接收方发回响应以后才发下一个数据包的通讯方式。
异步是指：发送方发出数据后，不等接收方发回响应，接着发送下个数据包的通讯方式。

7、请问multiprocessing模块中的Value、Array类的作用是什么？举例说明它们的使用场景

Value( typecode, arg1, … argN, lock )
在共享内容中常见ctypes对象。typecode要么是包含array模块使用的相同类型代码（如’i’,’d’等）的字符串，要么是来自ctypes模块的类型对象（如ctypes.c_int、ctypes.c_double等）。
所有额外的位置参数arg1, arg2 ….. argN将传递给指定类型的构造函数。lock是只能使用关键字调用的参数，如果把它置为True（默认值），将创建一个新的锁定来包含对值的访问。
如果传入一个现有锁定，比如Lock或RLock实例，该锁定将用于进行同步。如果v是Value创建的共享值的实例，便可使用v.value访问底层的值。例如，读取v.value将获取值，而赋值v.value将修改值。

RawValue( typecode, arg1, … ,argN)
同Value对象，但不存在锁定。

Array( typecode, initializer, lock )
在共享内存中创建ctypes数组。typecode描述了数组的内容，意义与Value()函数中的相同。initializer要么是设置数组初始大小的整数，要么是项目序列，其值和大小用于初始化数组。lock是只能使用关键字调用的参数，意义与Value()函数中相同。
如果a是Array创建的共享数组的实例，便可使用标准的python索引、切片和迭代操作访问它的内容，其中每种操作均由锁定进行同步。对于字节字符串，a还具有a.value属性，可以吧整个数组当做一个字符串进行访问。

RawArray(typecode, initializer )
同Array对象，但不存在锁定。当所编写的程序必须一次性操作大量的数组项时，如果同时使用这种数据类型和用于同步的单独锁定（如果需要的话），性能将得到极大的提升。
详见：https://blog.csdn.net/winterto1990/article/details/48106505

http://xiaorui.cc/2016/05/10/%E6%BA%90%E7%A0%81%E5%88%86%E6%9E%90multiprocessing%E7%9A%84value-array%E5%85%B1%E4%BA%AB%E5%86%85%E5%AD%98%E5%8E%9F%E7%90%86/

https://blog.csdn.net/qdx411324962/article/details/46810421

8、请问multiprocessing模块中的Manager类的作用是什么？与Value和Array类相比，Manager的优缺点是什么？

可以通过使用Value或者Array把数据存储在一个共享的内存表中；Manager()返回一个manager类型，控制一个server process，可以允许其它进程通过代理复制一些python objects 支持list,dict,Namespace,Lock,Semaphore,BoundedSemaphore,Condition,Event,Queue,Ｖalue,Array ；

Manager类的作用共享资源，manger的的优点是可以在poor进程池中使用，缺点是windows下环境下性能比较差，因为windows平台需要把Manager.list放在if name='main'下，而在实例化子进程时，必须把Manager对象传递给子进程，否则lists无法被共享，而这个过程会消耗巨大资源，因此性能很差。

参考：http://www.kaka-ace.com/python-multiprocessing-module-2-managers-first-profile/

https://blog.csdn.net/alvine008/article/details/24310939

9、写一个程序，包含十个线程，子线程必须等待主线程sleep 10秒钟之后才执行，并打印当前时间；

from threading import Thread
import time

def task(name):
    print(name, time.strftime('%Y-%m-%d %H:%M:%S',time.localtime()))

if __name__ == '__main__':
    time.sleep(10)
    for i in range(10):
        t = Thread(target=task, args=('线程 %s'%i, ))
        t.start()

10、写一个程序，包含十个线程，同时只能有五个子线程并行执行；

from threading import Thread,currentThread
from concurrent.futures import ThreadPoolExecutor
import time
import random

def task():
    print(currentThread().getName())
    time.sleep(random.randint(1,3))

if __name__ == "__main__":
    pool = ThreadPoolExecutor(5)
    for i in range(10):
        pool.submit(task)

11、写一个程序，要求用户输入用户名和密码，要求密码长度不少于6个字符，且必须以字母开头，如果密码合法，则将该密码使用md5算法加密后的十六进制概要值存入名为password.txt的文件，超过三次不合法则退出程序；

import hashlib
import json
import re

def func():
    count = 0
    while count < 3:
        username = input('username>>>:').strip()
        password = input('password>>>:').strip()
        if len(password) < 6 or not re.search('\A([a-z]|[A-Z])',password):
            count += 1
        else:
            obj = {'usename':username,'password':hashlib.md5(password.encode('utf-8')).hexdigest()}
            json.dump(obj,open('password.txt','a',encoding='utf-8'))
            break

if __name__ == "__main__":
    func()

12、写一个程序，使用socketserver模块，实现一个支持同时处理多个客户端请求的服务器，要求每次启动一个新线程处理客户端请求；

# server
import socketserver

class Handler(socketserver.BaseRequestHandler):   # 必须继承BaseRequestHandler
    def handle(self):    # 必须有handle方法
        print('new connection:',self.client_address)
        while True:
            try:
                data = self.request.recv(1024)
                if not data:break
                print('client data:',data.decode())
                self.request.send(data.upper())
            except Exception as e:
                print(e)
                break

if __name__ == "__main__":
    server = socketserver.ThreadingTCPServer(('127.0.0.1',8080),Handler)  # 实例化对象，实现多线程的socket
    server.serve_forever()  # 事件监听，并调用handler方法

# client
import socket
client = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
client.connect(('127.0.0.1',8080))
while True:
    msg = input(">>>:").strip()
    if not msg:continue
    client.send(msg.encode('utf-8'))
    data = client.recv(1024)
    print(data.decode('utf-8'))