python 学习总结2 多进程与协程

多进程：

我们什么时候需要多进程呢？我们知道python的多线程，实际不是真实的多线程，它同一时间在一个cpu执行一个任务，它通过上下文的切换来让我看起来是多并发的，

那么如果我们想要真正实现多个任务在多个cpu上同时执行，我们就需要多进程的性质来帮忙了（python的多线程不适合cpu密集型的任务，适合io密集型的任务）。

import multiprocessing
import threading

def thread_run():
    print(threading.get_ident())
def run():
    print("ok")
    t=threading.Thread(target=thread_run(),)
    t.start()
if __name__=="__main__":
    for i in range(10):
        p=multiprocessing.Process(target=run,)
        p.start()

上端代码体现了多进程用法其实就是在形式上与多线程是大同小异的！

通过学习多线程我们知道所有的线程都是有一个父线程

from multiprocessing import Process
import os


def info(title):
    print(title)
    #os.getppid 显示父线程的id
    print('parent process:', os.getppid())
    #os.getid显示子线程的id
    print('process id:', os.getpid())
    print("

")


def f(name):
    info('33[31;1mcalled from child process function f33[0m')
    print('hello', name)

if __name__ == '__main__':
    info('33[32;1mmain process line33[0m')
    p = Process(target=f, args=('bob',))
    p.start()

通过上面的代码，我们在主线程调用info显示了 主线程相对的 父线程，与子线程，而且我们发现父线程就是pycharm本身，其子线程作为相对于这段代码子线程的主线程，通过这段代码我们了解了Python的主线程与子线程之间的关系

多进程Queue：

那么我们如何实现主线程与子线程公用一块数据呢，也就是我们怎么才能实现两个进程之间的交互呢？看下文

多进程的Queue与线程中的队列是不同的，注意不能在多进程中调用线程的Queue因为每个线程都有各自占用一块内存，多线程是共享一块内存的，本质上是不同的

import multiprocessing
def f(qq):
    print("in child:",qq.qsize())
    qq.put([42, None, 'hello'])
if __name__ == '__main__':
    q = multiprocessing.Queue()
    q.put("test123")
    p = multiprocessing.Process(target=f, args=(q,))
    p.start()
    p.join()

多进程的pipes与manager：

我们的管道实际上就像socket 一样 ，其实就是相互收发的过程

from multiprocessing import Process, Pipe


def f(conn):
    #在子线程发送信息
    conn.send([42, None, 'hello from child'])
    conn.send([42, None, 'hello from child3'])
    print("",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())  # prints "[42, None, 'hello']"
    print("parent",parent_conn.recv())  # prints "[42, None, 'hello']"
    parent_conn.send(" from hshs")  # prints "[42, None, 'hello']"
    p.join()

下面是manager的用法 ，实际上都类似，体现了两个进程分享一些

from multiprocessing import Process, Manager
import os

def f(d, l):
    d[1] = '1'
    d['2'] = 2
    d["pid%s" %os.getpid()] = os.getpid()
    l.append(1)
    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)

数据进行修改
进程锁与进程池

我们为什么需要进程锁呢？答案就是当我们启动多个进程时会同时共享一块屏幕，为防止内容在显示的过程中发生乱套的现象引入了进程锁，下面演示了进程锁的用法

import multiprocessing
import os
def run():
    print("ok",os.getpid)
if __name__=="__main__":
    lock=multiprocessing.Lock()
    for i in range(10):
        multiprocessing.Process(target=run,).start()

进程池是干什么用的呢？ 通过进程池比如有十个进程，那么我让这一段时间内同时进行的只有三个，其它的都挂起，那么我们就用到了进程池

import multiprocessing
import os
import time

def run(i):
    time.sleep(2)
    print("every process come on!",os.getpid())
    return i+100
def bar(args):
    print("the process has done",args,os.getpid())

if __name__=="__main__":
    pool=multiprocessing.Pool(5)
    print("主进程的：",os.getpid())
    for i in range(10):
        #通过apply_async实现的是并行
        #callback是一个回调函数，当func中的函数执行完后主线程会自动调用一次callback中的函数
        pool.apply_async(func=run,args=(i,),callback=bar)
        #apply实际上实现的是异步也就是串行
        #pool.apply()
    #注意close与join的顺序，原因不理解
    pool.close()
    pool.join()  

协程：

协程又称为微线程，英文名coroutine ,换句话说，协程就是一个用户态的轻量级的线程

协程拥有自己的寄存器上下文和栈。协程调度切换时，将寄存器上下文和栈保存到其他地方，在切回来的时候，恢复先前保存的寄存器上下文和栈。因此：

协程能保留上一次调用时的状态（即所有局部状态的一个特定组合），每次过程重入时，就相当于进入上一次调用的状态，换种说法：进入上一次离开时所处逻辑流的位置。

协程的好处：

• 无需线程上下文切换的开销
• 无需原子操作锁定及同步的开销
  • 　　"原子操作(atomic operation)是不需要synchronized"，所谓原子操作是指不会被线程调度机制打断的操作；这种操作一旦开始，就一直运行到结束，中间不会有任何 context switch （切换到另一个线程）。原子操作可以是一个步骤，也可以是多个操作步骤，但是其顺序是不可以被打乱，或者切割掉只执行部分。视作整体是原子性的核心。
• 方便切换控制流，简化编程模型
• 高并发+高扩展性+低成本：一个CPU支持上万的协程都不是问题。所以很适合用于高并发处理。

缺点：

• 无法利用多核资源：协程的本质是个单线程,它不能同时将 单个CPU 的多个核用上,协程需要和进程配合才能运行在多CPU上.当然我们日常所编写的绝大部分应用都没有这个必要，除非是cpu密集型应用。
• 进行阻塞（Blocking）操作（如IO时）会阻塞掉整个程序

下面利用以前学过的yield演示类似协程的代码

import time
import queue


def consumer(name):
    print("--->starting eating baozi...")
    while True:
        #当后面执行.__next__之后才真正的给传递值
        new_baozi = yield
        print("[%s] is eating baozi %s" % (name, new_baozi))



def producer():
    con.__next__()
    con2.__next__()
    n = 0
    while n < 5:
        n += 1
        #开始将n传递给consumer
        con.send(n)
        con2.send(n)
        print("33[32;1m[producer]33[0m is making baozi %s" % n)


if __name__ == '__main__':
    con = consumer("c1")
    con2 = consumer("c2")
    p = producer()

我们可以从这个代码中看出，他在实现从producer与consumer之间的来回转换，我们的协程正是利用了这样的一个性质来实现了高效率的解决问题

我们下面通过greenlet实现手动的切换来实现协程的应用！！！（注意我们事先要导入一下gevent模块）

from greenlet import greenlet

def run1():
    print("1")
    #每次切换的时候我们都调用switch来进行切换
    gr2.switch()
    print("3")
    gr2.switch()

def run2():
    print("2")
    gr1.switch()
    print("4")

gr1=greenlet(run1)
gr2=greenlet(run2)
gr1.switch()

那么如何实现自动切换呢？我们就用到了Gevent来实现自动的切换

import gevent

def foo():
    print('1')
    gevent.sleep(0)
    print('2')
def bar():
    print('3')
    gevent.sleep(1)
    print('4')
def func3():
    print("5 ")
    gevent.sleep(2)
    print("6 ")

#实际我认为上下文的切换在系统就是上下按顺序执行，他看到这个函数在wait就会自动向下找一下个，以此类推，来回往复
#从而实现了高效率的执行任务
gevent.joinall([
    gevent.spawn(foo), #生成，
    gevent.spawn(bar),
    gevent.spawn(func3),
])

 我们来用协程实现一个简单的网络爬虫

import gevent,time
from urllib import request
from gevent import monkey
#这个可以让urllib知道谁在进行io操作
monkey.patch_all()

def f(url):
    print("wo get:%s"%url)
    resp=request.urlopen(url)
    data=resp.read()
    print("%s recivied from %s"%(len(data),data))
start_time=time.time()
#能将两个操作变成协程
gevent.joinall([
    gevent.spawn(f, 'https://www.python.org/'),
    gevent.spawn(f, 'https://www.yahoo.com/'),
])
print("异步cost",time.time() - start_time)

那么我们如何用协程实现一个socket这样我们就可以大并发的处理

import socket
#客户端和普通的没什么两样
s=socket.socket()
s.connect(("localhost",22))
while True:
    msg=input("输入发送消息")
    s.sendall(msg.encode())
    data=s.recv(1024)
    print("recv",data.decode())
s.close()

服务器端利用了gevent实现了大并发：

import socket
import gevent
from gevent import monkey

monkey.patch_all()

def server(port):
    s=socket.socket()
    s.bind(("localhost",port))
    s.listen(500)
    while True:
        conn,addr=s.accept()
        gevent.spawn(handle_requset,conn)
def handle_requset(conn):
    try:
        while True:
            data=conn.recv(1024)
            print("recv:",data.decode())
            conn.send(data)
            if not data:
                conn.shutdown(socket.SHUT_WR)
    except Exception as ex:
        print("error：",ex)
    finally:
        conn.close()
if __name__=="__main__":
    server(22)

我们在从前的学习过程中学习过

#这是多线程的模式

server=socketserver.threadingTCPServer((host,port),MyTCPHandle)

#这是多进程的模式，这种我们通常不会去用，严重损耗资源

server=socketserver.ForkingTCPServer((host,port),MyTCPHandle)

论事件驱动与异步IO

在UI编程中，常常要对鼠标点击进行相应，首先如何获得鼠标点击呢？
方式一：创建一个线程，该线程一直循环检测是否有鼠标点击，那么这个方式有以下几个缺点：
1. CPU资源浪费，可能鼠标点击的频率非常小，但是扫描线程还是会一直循环检测，这会造成很多的CPU资源浪费；如果扫描鼠标点击的接口是阻塞的呢？
2. 如果是堵塞的，又会出现下面这样的问题，如果我们不但要扫描鼠标点击，还要扫描键盘是否按下，由于扫描鼠标时被堵塞了，那么可能永远不会去扫描键盘；
3. 如果一个循环需要扫描的设备非常多，这又会引来响应时间的问题；
所以，该方式是非常不好的。

方式二：就是事件驱动模型
目前大部分的UI编程都是事件驱动模型，如很多UI平台都会提供onClick()事件，这个事件就代表鼠标按下事件。事件驱动模型大体思路如下：
1. 有一个事件（消息）队列；
2. 鼠标按下时，往这个队列中增加一个点击事件（消息）；
3. 有个循环，不断从队列取出事件，根据不同的事件，调用不同的函数，如onClick()、onKeyDown()等；
4. 事件（消息）一般都各自保存各自的处理函数指针，这样，每个消息都有独立的处理函数；

下面这张图体现了单线程，多线程，事件驱动编程的三者效率对比

进程的阻塞

正在执行的进程，由于期待的某些事件未发生，如请求系统资源失败、等待某种操作的完成、新数据尚未到达或无新工作做等，则由系统自动执行阻塞原语(Block)，使自己由运行状态变为阻塞状态。可见，进程的阻塞是进程自身的一种主动行为，也因此只有处于运行态的进程（获得CPU），才可能将其转为阻塞状态。当进程进入阻塞状态，是不占用CPU资源的。

文件描述符fd

文件描述符（File descriptor）是计算机科学中的一个术语，是一个用于表述指向文件的引用的抽象化概念。

文件描述符在形式上是一个非负整数。实际上，它是一个索引值，指向内核为每一个进程所维护的该进程打开文件的记录表。当程序打开一个现有文件或者创建一个新文件时，内核向进程返回一个文件描述符。在程序设计中，一些涉及底层的程序编写往往会围绕着文件描述符展开。但是文件描述符这一概念往往只适用于UNIX、Linux这样的操作系统。

缓存 I/O

缓存 I/O 又被称作标准 I/O，大多数文件系统的默认 I/O 操作都是缓存 I/O。在 Linux 的缓存 I/O 机制中，操作系统会将 I/O 的数据缓存在文件系统的页缓存（ page cache ）中，也就是说，数据会先被拷贝到操作系统内核的缓冲区中，然后才会从操作系统内核的缓冲区拷贝到应用程序的地址空间。

缓存 I/O 的缺点：
数据在传输过程中需要在应用程序地址空间和内核进行多次数据拷贝操作，这些数据拷贝操作所带来的 CPU 以及内存开销是非常大的。

 刚才说了，对于一次IO访问（以read举例），数据会先被拷贝到操作系统内核的缓冲区中，然后才会从操作系统内核的缓冲区拷贝到应用程序的地址空间。所以说，当一个read操作发生时，它会经历两个阶段：
1. 等待数据准备 (Waiting for the data to be ready)
2. 将数据从内核拷贝到进程中 (Copying the data from the kernel to the process)

正式因为这两个阶段，linux系统产生了下面五种网络模式的方案。
- 阻塞 I/O（blocking IO）
- 非阻塞 I/O（nonblocking IO）
- I/O 多路复用（ IO multiplexing）
- 信号驱动 I/O（ signal driven IO）
- 异步 I/O（asynchronous IO）

注：由于signal driven IO在实际中并不常用，所以我这只提及剩下的四种IO Model。

阻塞 I/O（blocking IO）

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

所以他的特点就是两个过程都被阻塞

非阻塞 I/O（nonblocking IO）

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

对于非阻塞的IO阻塞在第二个过程中

I/O 多路复用（ IO multiplexing）

IO multiplexing就是我们说的select，poll，epoll，有些地方也称这种IO方式为event driven IO。select/epoll的好处就在于单个process就可以同时处理多个网络连接的IO。它的基本原理就是select，poll，epoll这个function会不断的轮询所负责的所有socket，当某个socket有数据到达了，就通知用户进程。

多路复用相对与阻塞I/O的来说，可以有多个socket，只要有一个人的数据发过来就会返回一个数据

异步 I/O（asynchronous IO）

inux下的asynchronous IO其实用得很少。先看一下它的流程：

很形象的比喻他就是他一点都不卡顿为什么呢？这就像办理业务，你先去把你信息填好，等到轮到你的时候 它自动处理好后将信息给你发过去，不会有什么等待的过程！！！

 下面是多路复用中select的实现过程

import socket
import select
server=socket.socket()
server.bind(("localhost",22))
server.listen(1000)
#我们先将代码设置成非阻塞模式
server.setblocking(False)
#就是所有的来的链接，活跃的项目的储存处，我们知道多路复用是可以支持多个连接同时应用所以我们用上了列表存储
inputs=[server,]
#暂时没学到outputs的解释
outputs=[]
while True:
    readable,writealbe,exceptional=select.select(inputs,outputs,inputs)
    for r in readable:
        if r is server:
            conn,addr=server.accept()
            print("new addr:",addr)
            inputs.append(conn)
            print(r)
        else:
            data=r.recv(1024)
            print(r)
            print(data)

 我们要知道一点select，poll,epoll都属于很基层的东西，我们暂时先不需研究过深

因为selectors这个模块已经帮我将底层的封装好了 如果电脑支持的是select的话那么就有select，有epoll就会用epoll

服务器端：

import selectors
import socket

sel=selectors.DefaultSelector()

def accept(sock,mask):
    conn,addr=sock.accept()
    print('accepted', conn, 'from', addr)
    print(mask)
    conn.setblocking(False)
    sel.register(conn,selectors.EVENT_READ,read)


def read(conn,mask):
    data=conn.recv(1024)
    if data:
        print(data,mask)
        conn.send(data)
    else:
        print("closing",conn)
        sel.unregister(conn)
        conn.close()



server=socket.socket()
server.bind(("localhost",23))
server.listen(1000)
server.setblocking(False)#设置为非阻塞模式
sel.register(server,selectors.EVENT_READ,accept)

while True:
    events=sel.select()
    for key,mask in events:
        callback=key.data#相当于callback=accept
        callback(key.fileobj,mask)#调用accept keyfileobj相当于一个实例作为形参传进去

客户端：

import socket
import sys

messages=[b"woaini",
          b"niaishei",
          b"ainima"
          ]


socks=[socket.socket() for i in range (290)]
#这样我们就同时启动了290是个client链接

for s in socks:
    s.connect(("localhost",23))

for message in messages:
    for i in socks:
        i.send(message)
    for s in socks:
        data=s.recv(1024)
        print("returning :",data)
        if not data:
            print("closing socket")

 上面服务器端与客户端充分体现出了 多路复用的强大之处！！如果我们用Linux的话会支持更多的并发！！！