Python异步非阻塞IO多路复用Select/Poll/Epoll使用，线程，进程，协程

1.使用select模拟socketserver伪并发处理客户端请求，代码如下：

import socket

import select

sk = socket.socket()

sk.bind(('127.0.0.1', 9999,))

sk.listen(5)

inputs = [sk,]

outputs = []

messages = {}

# del messages[白宇]

# messages = {

#    白宇：[消息1，消息2，]

#    吴文煜：[消息1，消息2，]

# }

while True:

    rlist,wlist,elist, = select.select(inputs, outputs,[sk,],1)

    print(len(inputs),len(rlist),len(wlist), len(outputs))

    # 监听sk（服务器端）对象，如果sk对象发生变化，表示有客户端来连接了，此时rlist值为【sk】

    # 监听conn对象，如果conn发生变化，表示客户端有新消息发送过来了，此时rlist的之为 【客户端】

    # rlist = 【吴文煜，】

    # rlist = 【张磊，白宇，】

    # rlist = [sk,]

    # wlist 发过消息的对象追加到添加到此列表中

    # elist socket出现异常情况，追加到此列表中

    for r in rlist:

        if r == sk:

            # 新客户来连接

            conn, address = r.accept()

            # conn是什么？其实socket对象

            inputs.append(conn)

            messages[conn] = []

            conn.sendall(bytes('hello', encoding='utf-8'))

        else:

            # 有人给我发消息了

            print('=======')

            try:

                ret = r.recv(1024)

                # r.sendall(ret)

                if not ret:

                    raise Exception('断开连接')

                else:

                    outputs.append(r)

                    messages[r].append(ret)

            except Exception as e:

                inputs.remove(r)

                del messages[r]

    # 所有给我发过消息的人

    for w in wlist:

        msg = messages[w].pop()

        resp = msg + bytes('response', encoding='utf-8')

        w.sendall(resp)

        outputs.remove(w)

# rlist = [sk,],rlist=[sk1,],rlist = [sk1,sk2]

# rlist = []

socket_select_server

import socket

sk = socket.socket()

sk.connect(("127.0.0.1", 9999,))

data = sk.recv(1024)

print(data)

while True:

    inp = input(">>>")

    sk.sendall(bytes(inp,encoding='utf-8'))

    print(sk.recv(1024))

sk.close()

client

有许多封装好的异步非阻塞IO多路复用框架，底层在linux基于最新的epoll实现，为了更好的使用，了解其底层原理还是有必要的。
下面记录下分别基于Select/Poll/Epoll的echo server实现。
Python Select Server，可监控事件数量有限制：

#!/usr/bin/python

# -*- coding: utf-8 -*-

import select

import socket

import Queue

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

server.setblocking(False)

server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR , 1)

server_address= ('192.168.1.5',8080)

server.bind(server_address)

server.listen(10)

#select轮询等待读socket集合

inputs = [server]

#select轮询等待写socket集合

outputs = []

message_queues = {}

#select超时时间

timeout = 20

while True:

    print "等待活动连接......"

    readable , writable , exceptional = select.select(inputs, outputs, inputs, timeout)

    if not (readable or writable or exceptional) :

        print "select超时无活动连接，重新select...... "

        continue;

    #循环可读事件

    for s in readable :

        #如果是server监听的socket

        if s is server:

            #同意连接

            connection, client_address = s.accept()

            print "新连接： ", client_address

            connection.setblocking(0)

            #将连接加入到select可读事件队列

            inputs.append(connection)

            #新建连接为key的字典，写回读取到的消息

            message_queues[connection] = Queue.Queue()

        else:

            #不是本机监听就是客户端发来的消息

            data = s.recv(1024)

            if data :

                print "收到数据：" , data , "客户端：",s.getpeername()

                message_queues[s].put(data)

                if s not in outputs:

                    #将读取到的socket加入到可写事件队列

                    outputs.append(s)

            else:

                #空白消息，关闭连接

                print "关闭连接：", client_address

                if s in outputs :

                    outputs.remove(s)

                inputs.remove(s)

                s.close()

                del message_queues[s]

    for s in writable:

        try:

            msg = message_queues[s].get_nowait()

        except Queue.Empty:

            print "连接：" , s.getpeername() , '消息队列为空'

            outputs.remove(s)

        else:

            print "发送数据：" , msg , "到", s.getpeername()

            s.send(msg)

    for s in exceptional:

        print "异常连接：", s.getpeername()

        inputs.remove(s)

        if s in outputs:

            outputs.remove(s)

        s.close()

        del message_queues[s]

Python Select Code

Python Poll Server，Select升级版，无可监控事件数量限制，还是要轮询所有事件：

#!/usr/bin/python

# -*- coding: utf-8 -*-

import socket

import select

import Queue

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

server.setblocking(False)

server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

server_address = ("192.168.1.5", 8080)

server.bind(server_address)

server.listen(5)

print "服务器启动成功，监听IP：" , server_address

message_queues = {}

#超时，毫秒

timeout = 5000

#监听哪些事件

READ_ONLY = ( select.POLLIN | select.POLLPRI | select.POLLHUP | select.POLLERR)

READ_WRITE = (READ_ONLY|select.POLLOUT)

#新建轮询事件对象

poller = select.poll()

#注册本机监听socket到等待可读事件事件集合

poller.register(server,READ_ONLY)

#文件描述符到socket映射

fd_to_socket = {server.fileno():server,}

while True:

    print "等待活动连接......"

    #轮询注册的事件集合

    events = poller.poll(timeout)

    if not events:

      print "poll超时，无活动连接，重新poll......"

      continue

    print "有" , len(events), "个新事件，开始处理......"

    for fd ,flag in events:

        s = fd_to_socket[fd]

        #可读事件

        if flag & (select.POLLIN | select.POLLPRI) :

            if s is server :

                #如果socket是监听的server代表有新连接

                connection , client_address = s.accept()

                print "新连接：" , client_address

                connection.setblocking(False)

                fd_to_socket[connection.fileno()] = connection

                #加入到等待读事件集合

                poller.register(connection,READ_ONLY)

                message_queues[connection] = Queue.Queue()

            else :

                #接收客户端发送的数据

                data = s.recv(1024)

                if data:

                    print "收到数据：" , data , "客户端：" , s.getpeername()

                    message_queues[s].put(data)

                    #修改读取到消息的连接到等待写事件集合

                    poller.modify(s,READ_WRITE)

                else :

                    # Close the connection

                    print " closing" , s.getpeername()

                    # Stop listening for input on the connection

                    poller.unregister(s)

                    s.close()

                    del message_queues[s]

        #连接关闭事件

        elif flag & select.POLLHUP :

            print " Closing ", s.getpeername() ,"(HUP)"

            poller.unregister(s)

            s.close()

        #可写事件

        elif flag & select.POLLOUT :

            try:

                msg = message_queues[s].get_nowait()

            except Queue.Empty:

                print s.getpeername() , " queue empty"

                poller.modify(s,READ_ONLY)

            else :

                print "发送数据：" , data , "客户端：" , s.getpeername()

                s.send(msg)

        #异常事件

        elif flag & select.POLLERR:

            print " exception on" , s.getpeername()

            poller.unregister(s)

            s.close()

            del message_queues[s]

Python Poll server

Python Epoll Server，基于回调的事件通知模式，轻松管理大量连接：

#!/usr/bin/python

# -*- coding: utf-8 -*-

import socket, select

import Queue

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

serversocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

server_address = ("192.168.1.5", 8080)

serversocket.bind(server_address)

serversocket.listen(1)

print "服务器启动成功，监听IP：" , server_address

serversocket.setblocking(0)

timeout = 10

#新建epoll事件对象，后续要监控的事件添加到其中

epoll = select.epoll()

#添加服务器监听fd到等待读事件集合

epoll.register(serversocket.fileno(), select.EPOLLIN)

message_queues = {}

fd_to_socket = {serversocket.fileno():serversocket,}

while True:

  print "等待活动连接......"

  #轮询注册的事件集合

  events = epoll.poll(timeout)

  if not events:

     print "epoll超时无活动连接，重新轮询......"

     continue

  print "有" , len(events), "个新事件，开始处理......"

  for fd, event in events:

     socket = fd_to_socket[fd]

     #可读事件

     if event & select.EPOLLIN:

         #如果活动socket为服务器所监听，有新连接

         if socket == serversocket:

            connection, address = serversocket.accept()

            print "新连接：" , address

            connection.setblocking(0)

            #注册新连接fd到待读事件集合

            epoll.register(connection.fileno(), select.EPOLLIN)

            fd_to_socket[connection.fileno()] = connection

            message_queues[connection] = Queue.Queue()

         #否则为客户端发送的数据

         else:

            data = socket.recv(1024)

            if data:

               print "收到数据：" , data , "客户端：" , socket.getpeername()

               message_queues[socket].put(data)

               #修改读取到消息的连接到等待写事件集合

               epoll.modify(fd, select.EPOLLOUT)

     #可写事件

     elif event & select.EPOLLOUT:

        try:

           msg = message_queues[socket].get_nowait()

        except Queue.Empty:

           print socket.getpeername() , " queue empty"

           epoll.modify(fd, select.EPOLLIN)

        else :

           print "发送数据：" , data , "客户端：" , socket.getpeername()

           socket.send(msg)

     #关闭事件

     elif event & select.EPOLLHUP:

        epoll.unregister(fd)

        fd_to_socket[fd].close()

        del fd_to_socket[fd]

epoll.unregister(serversocket.fileno())

epoll.close()

serversocket.close()

Python Epoll server

初始化进程、线程与协成的概念

什么是进程？

　进程，是计算机中的程序关于某数据集合上的一次运行活动，是系统进行资源分配和调度的基本单位，是操作系统结构的基础。前面的话我也没懂，用非官方的白话来解释就是——执行中的程序是进程，比如qq不是进程，但是当我们双击qq开始使用它的时候，它就变成了一个进程。我们写的python程序，只有当我们执行它的时候，它才是进程。我们正在执行的IE浏览器，QQ，pycharm都是进程，从操作系统的角度来讲，每一个进程都有它自己的内存空间，进程之间的内存是独立的。

什么是线程？

　　线程，有时被称为轻量级进程，是程序执行流的最小单元。我们可以理解为，线程是属于进程的，我们平时写的简单程序，是单线程的，多线程和单线程的区别在于多线程可以同时处理多个任务，这时候我们可以理解为多线程和多进程是一样的，我可以在我的进程中开启一个线程放音乐，也可以开启另外的线程聊qq，但是进程之间的内存独立，而属于同一个进程多个线程之间的内存是共享的，多个线程可以直接对它们所在进程的内存数据进行读写并在线程间进行交换。

进程与线程之间的关系

先推荐一个链接，这篇文章用漫画的形式讲解了进程与线程的关系：http://www.ruanyifeng.com/blog/2013/04/processes_and_threads.html

在python界一直有着一个古老的传说，那就是python的多线程是鸡肋，那么这个传说的信度到底有多少呢？如果我们的代码是CPU密集型（涉及到大量的计算），多个线程的代码很有可能是线性执行的，所以这种情况下多线程是鸡肋，效率可能还不如单线程，因为有context switch（其实就是线程之间的切换和线程的创建等等都是需要消耗时间的）；但是：如果是IO密集型，多线程可以明显提高效率。例如制作爬虫，绝大多数时间爬虫是在等待socket返回数据。这个时候C代码里是有release GIL的，最终结果是某个线程等待IO的时候其他线程可以继续执行。

　　那么，为什么我们大python会这么不智能呢？我们都知道，python是一种解释性语言，在python执行的过程中，需要解释器一边解释一边执行，我们之前也介绍了，同一个进程的线程之间内存共享，那么就会出现内存资源的安全问题，python为了线程安全，就设置了全局解释器锁机制，既一个进程中同时只能有一个线程访问cpu。作为解释型语言，python能引入多线程的概念就已经非常不易了，目前看到的资料php和perl等多线程机制都是不健全的。解释型语言做多线程的艰难程度可以想见。。。具体下面的链接推荐：python的最难问题。

　　正是由于python多线程的缺陷，我们在这里需要引入协成的概念。

什么是协程？

　　协程是一种用户态的轻量级线程。如果说多进程对于多CPU，多线程对应多核CPU，那么事件驱动和协程则是在充分挖掘不断提高性能的单核CPU的潜力。我们既可以利用异步优势，又可以避免反复系统调用，还有进程切换造成的开销，这就是协程。协程也是单线程，但是它能让原来要使用异步+回调方式写的非人类代码,可以用看似同步的方式写出来。它是实现推拉互动的所谓非抢占式协作的关键。对于python来说，由于python多线程中全局解释器导致的同时只能有一个线程访问cpu，所以对协程需求就相比于其他语言更为紧迫。

进程、线程与协程

　　从硬件发展来看，从最初的单核单CPU，到单核多CPU，多核多CPU，似乎已经到了极限了，但是单核CPU性能却还在不断提升。server端也在不断的发展变化。如果将程序分为IO密集型应用和CPU密集型应用，二者的server的发展如下：

　　　　IO密集型应用: 多进程->多线程->事件驱动->协程
　　　　CPU密集型应用:多进程-->多线程

　　调度和切换的时间：进程 > 线程 > 协程

偷懒的同学看这里→_→：不需要实现复杂的内存共享且需利用多cpu，用多进程；实现复杂的内存共享及IO密集型应用：多线程或协程；实现复杂的内存共享及CPU密集型应用：协程