python3之协程

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2018/1/24 10:46
# @Author  : Py.qi
# @File    : yield_xiecheng.py
# @Software: PyCharm

def consumer(name):
    print(‘开始吃包子...‘)
    while True:
        print(‘\033[31;1m[consumer]%s需要包子\033[0m‘%name)
        bone = yield   #接收send发送的数据
        print(‘\033[31;1m[%s]吃了%s个包子\033[0m‘%(name,bone))
def producer(obj1):
    obj1.send(None)   #必须先发送None
    for i in range(3):
        print(‘\033[32;1m[producer]\033[0m正在做%s个包子‘%i)
        obj1.send(i)


if __name__ == ‘__main__‘:
    con1 = consumer(‘消费者A‘)  #创建消费者对象
    producer(con1)

#output：
开始吃包子...
[consumer]消费者A需要包子
[producer]正在做0个包子
[消费者A]吃了0个包子
[consumer]消费者A需要包子
[producer]正在做1个包子
[消费者A]吃了1个包子
[consumer]消费者A需要包子
[producer]正在做2个包子
[消费者A]吃了2个包子
[consumer]消费者A需要包子

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2018/1/24 15:25
# @Author  : Py.qi
# @File    : greenlet_now.py
# @Software: PyCharm
import greenlet

def A():
    print(‘a.....‘)
    g2.switch()  #切换至B
    print(‘a....2‘)
    g2.switch()
def B():
    print(‘b.....‘)
    g1.switch()  #切换至A
    print(‘b....2‘)

g1 = greenlet.greenlet(A) #启动一个线程
g2 = greenlet.greenlet(B)
g1.switch()

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2018/1/24 15:59
# @Author  : Py.qi
# @File    : gevent_noe.py
# @Software: PyCharm
import gevent

def foo():
    print(‘running in foo‘)
    gevent.sleep(2)
    print(‘com back from bar in to foo‘)
def bar():
    print(‘running in bar‘)
    gevent.sleep(2)
    print(‘com back from foo in to bar‘)

gevent.joinall([      #创建线程并行执行程序，碰到IO就切换
    gevent.spawn(foo),
    gevent.spawn(bar),
])

import gevent
def task(pid):
    gevent.sleep(1)
    print(‘task %s done‘%pid)

def synchronous():  #同步一个线程执行函数
    for i in range(1,10):
        task(i)
def asynchronous(): #异步一个线程执行函数
    threads = [gevent.spawn(task,i) for i in range(10)]
    gevent.joinall(threads)

print(‘synchronous:‘)
synchronous()   #同步执行时要等待执行完后再执行
print(‘asynchronous:‘)
asynchronous()  #异步时遇到等待则会切换执行

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2018/1/24 17:00
# @Author  : Py.qi
# @File    : gevent_urllib.py
# @Software: PyCharm

from urllib import request
import gevent,time
from gevent import monkey

monkey.patch_all()   #将程序中所有IO操作做上标记使程序非阻塞状态
def url_request(url):
    print(‘get:%s‘%url)
    resp = request.urlopen(url)
    data = resp.read()
    print(‘%s bytes received from %s‘%(len(data),url))

async_time_start = time.time() #开始时间
gevent.joinall([
    gevent.spawn(url_request,‘https://www.python.org/‘),
    gevent.spawn(url_request,‘https://www.nginx.org/‘),
    gevent.spawn(url_request,‘https://www.ibm.com‘),
])
print(‘haoshi:‘,time.time()-async_time_start) #总用时

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2018/1/24 17:22
# @Author  : Py.qi
# @File    : gevent_sock.py
# @Software: PyCharm
import socket,gevent
from gevent import monkey
monkey.patch_all()

def server_sock(port):
    s = socket.socket()
    s.bind((‘‘,port))
    s.listen(10)
    while True:
        conn,addr = s.accept()
        gevent.spawn(handle_request,conn)
def handle_request(conn):
    try:
        while True:
            data = conn.recv(1024)
            if not data: conn.shutdown(socket.SHUT_WR)
            print(‘recv:‘,data.decode())
            conn.send(data)
    except Exception as ex:
        print(ex)
    finally:
        conn.close()

if __name__ == ‘__main__‘:
    server_sock(8888)




#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2018/1/24 17:35
# @Author  : Py.qi
# @File    : gevent_sockclient.py
# @Software: PyCharm

import socket

HOST = ‘localhost‘  # The remote host
PORT = 8888  # The same port as used by the server
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect((HOST, PORT))
while True:
    #msg = bytes(input(">>:"), encoding="utf8")
    for i in range(50):
        s.send(‘dddd‘.encode())
        data = s.recv(1024)
    # print(data)

        print(‘Received‘, repr(data))
    s.close()

struct pollfd {
    int fd; /*file descriptor */
    short events; /* requested events to watch */
    short revents; /* returned events witnessed */
};

int epoll_create(int size)；//创建一个epoll的句柄，size用来告诉内核这个监听的数目一共有多大
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event)；
int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout);

#!/usr/bin/python
#Author:sean
 
import select
import socket
import queue
 
server = socket.socket()
HOST = ‘localhost‘
PORT = 8080
print("start up %s on port: %s",% (HOST,PORT))
server.bind((HOST,PORT))
server.listen()
 
server.setblocking(False)   #不阻塞
 
msg_dic_queue = {}    #这是一个队列字典，存放要返回给客户端的数据
 
inputs = [server]   #inputs里存放要让内核监测的连接，这里的server是指监测server本身的连接状态
#inputs = [server,conn]
outputs = []    #outputs里存放要返回给客户端的数据连接对象
 
while True:
    print("waiting for next connect...")
    readable,writeable,exceptional = select.select(inputs,outputs,inputs)   #如果没有任何fd就绪，程序就会一直阻塞在这里
    # print(readable,writeable,exceptional)
    for r in readable:  #处理活跃的连接，每个r就是一个socket连接对象
        if r is server: #代表来了一个新连接
            conn,client_addr = server.accept()
            print("arrived a new connect: ",client_addr)
            conn.setblocking(False)
            inputs.append(conn) #因为这个新建立的连接还没发数据来，现在就接收的话，程序就报异常了
            #所以要想实现这个客户端发数据来时server端能知道，就需要让select再监测这个conn
            msg_dic_queue[conn] = queue.Queue()   #初始化一个队列，后面存要返回给客户端的数据
        else:   #r不是server的话就代表是一个与客户端建立的文件描述符了
            #客户端的数据过来了，在这里接收
            data = r.recv(1024)
            if data:
                print("received data from [%s]: "% r.getpeername()[0],data)
                msg_dic_queue[r].put(data)  #收到的数据先放到队列字典里，之后再返回给客户端
                if r not in outputs:
                    outputs.append(r)   #放入返回的连接队列里。为了不影响处理与其它客户端的连接，这里不立刻返回数据给客户端
            else:   #如果收不到data就代表客户端已经断开了
                print("Client is disconnect",r)
                if r in outputs:
                    outputs.remove(r)   #清理已断开的连接
                inputs.remove(r)
                del msg_dic_queue[r]
 
    for w in writeable: #处理要返回给客户端的连接列表
        try:
            next_msg = msg_dic_queue[w].get_nowait()
        except queue.Empty:
            print("client [%s]"% w.getpeername()[0],"queue is empty...")
            outputs.remove(w)   #确保下次循环时writeable不返回已经处理完的连接
        else:
            print("sending message to [%s]"% w.getpeername()[0],next_msg)
            w.send(next_msg)    #返回给客户端源数据
 
    for e in exceptional:   #处理异常连接
        if e in outputs:
            outputs.remove(e)
        inputs.remove(e)
        del msg_dic_queue[e]

#!/usr/bin/python
#Author:sean
 
import socket
 
msgs = [ b‘This is the message. ‘,
             b‘It will be sent ‘,
             b‘in parts.‘,
             ]
SERVER_ADDRESS = ‘localhost‘
SERVER_PORT = 8080
 
# Create a few TCP/IP socket
socks = [ socket.socket(socket.AF_INET, socket.SOCK_STREAM) for i in range(500) ]
 
# Connect the socket to the port where the server is listening
print(‘connecting to %s port %s‘ % (SERVER_ADDRESS,SERVER_PORT))
for s in socks:
    s.connect((SERVER_ADDRESS,SERVER_PORT))
 
for message in msgs:
 
    # Send messages on both sockets
    for s in socks:
        print(‘%s: sending "%s"‘ % (s.getsockname(), message) )
        s.send(message)
 
    # Read responses on both sockets
    for s in socks:
        data = s.recv(1024)
        print( ‘%s: received "%s"‘ % (s.getsockname(), data) )
        if not data:
            print(sys.stderr, ‘closing socket‘, s.getsockname() )

#!/usr/bin/python
#Author:sean
 
import socket, logging
import select, errno
 
logger = logging.getLogger("network-server")
 
def InitLog():
    logger.setLevel(logging.DEBUG)
 
    fh = logging.FileHandler("network-server.log")
    fh.setLevel(logging.DEBUG)
    ch = logging.StreamHandler()
    ch.setLevel(logging.ERROR)
 
    formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
    ch.setFormatter(formatter)
    fh.setFormatter(formatter)
 
    logger.addHandler(fh)
    logger.addHandler(ch)
 
 
if __name__ == "__main__":
    InitLog()
 
    try:
        # 创建 TCP socket 作为监听 socket
        listen_fd = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
    except socket.error as  msg:
        logger.error("create socket failed")
 
    try:
        # 设置 SO_REUSEADDR 选项
        listen_fd.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    except socket.error as  msg:
        logger.error("setsocketopt SO_REUSEADDR failed")
 
    try:
        # 进行 bind -- 此处未指定 ip 地址，即 bind 了全部网卡 ip 上
        listen_fd.bind((‘‘, 8008))
    except socket.error as  msg:
        logger.error("bind failed")
 
    try:
        # 设置 listen 的 backlog 数
        listen_fd.listen(10)
    except socket.error as  msg:
        logger.error(msg)
 
    try:
        # 创建 epoll 句柄
        epoll_fd = select.epoll()
        # 向 epoll 句柄中注册 监听 socket 的 可读 事件
        epoll_fd.register(listen_fd.fileno(), select.EPOLLIN)
    except select.error as  msg:
        logger.error(msg)
 
    connections = {}
    addresses = {}
    datalist = {}
    while True:
        # epoll 进行 fd 扫描的地方 -- 未指定超时时间则为阻塞等待
        epoll_list = epoll_fd.poll()
 
        for fd, events in epoll_list:
            # 若为监听 fd 被激活
            if fd == listen_fd.fileno():
                # 进行 accept -- 获得连接上来 client 的 ip 和 port，以及 socket 句柄
                conn, addr = listen_fd.accept()
                logger.debug("accept connection from %s, %d, fd = %d" % (addr[0], addr[1], conn.fileno()))
                # 将连接 socket 设置为 非阻塞
                conn.setblocking(0)
                # 向 epoll 句柄中注册 连接 socket 的 可读 事件
                epoll_fd.register(conn.fileno(), select.EPOLLIN | select.EPOLLET)
                # 将 conn 和 addr 信息分别保存起来
                connections[conn.fileno()] = conn
                addresses[conn.fileno()] = addr
            elif select.EPOLLIN & events:
                # 有 可读 事件激活
                datas = ‘‘
                while True:
                    try:
                        # 从激活 fd 上 recv 10 字节数据
                        data = connections[fd].recv(10)
                        # 若当前没有接收到数据，并且之前的累计数据也没有
                        if not data and not datas:
                            # 从 epoll 句柄中移除该 连接 fd
                            epoll_fd.unregister(fd)
                            # server 侧主动关闭该 连接 fd
                            connections[fd].close()
                            logger.debug("%s, %d closed" % (addresses[fd][0], addresses[fd][1]))
                            break
                        else:
                            # 将接收到的数据拼接保存在 datas 中
                            datas += data
                    except socket.error as  msg:
                        # 在 非阻塞 socket 上进行 recv 需要处理 读穿 的情况
                        # 这里实际上是利用 读穿 出 异常 的方式跳到这里进行后续处理
                        if msg.errno == errno.EAGAIN:
                            logger.debug("%s receive %s" % (fd, datas))
                            # 将已接收数据保存起来
                            datalist[fd] = datas
                            # 更新 epoll 句柄中连接d 注册事件为 可写
                            epoll_fd.modify(fd, select.EPOLLET | select.EPOLLOUT)
                            break
                        else:
                            # 出错处理
                            epoll_fd.unregister(fd)
                            connections[fd].close()
                            logger.error(msg)
                            break
            elif select.EPOLLHUP & events:
                # 有 HUP 事件激活
                epoll_fd.unregister(fd)
                connections[fd].close()
                logger.debug("%s, %d closed" % (addresses[fd][0], addresses[fd][1]))
            elif select.EPOLLOUT & events:
                # 有 可写 事件激活
                sendLen = 0
                # 通过 while 循环确保将 buf 中的数据全部发送出去
                while True:
                    # 将之前收到的数据发回 client -- 通过 sendLen 来控制发送位置
                    sendLen += connections[fd].send(datalist[fd][sendLen:])
                    # 在全部发送完毕后退出 while 循环
                    if sendLen == len(datalist[fd]):
                        break
                # 更新 epoll 句柄中连接 fd 注册事件为 可读
                epoll_fd.modify(fd, select.EPOLLIN | select.EPOLLET)
            else:
                # 其他 epoll 事件不进行处理
                continue

#!/usr/bin/python
#Author:sean
 
import selectors
import socket
#selectors模块默认会用epoll，如果你的系统中没有epoll(比如windows)则会自动使用select
sel = selectors.DefaultSelector()   #生成一个select对象
 
def accept(sock, mask):
    conn, addr = sock.accept()  # Should be ready
    print(‘accepted‘, conn, ‘from‘, addr)
    conn.setblocking(False) #设定非阻塞
    sel.register(conn, selectors.EVENT_READ, read)  #新连接注册read回调函数
 
def read(conn, mask):
    data = conn.recv(1024)  # Should be ready
    if data:
        print(‘echoing‘, repr(data), ‘to‘, conn)
        conn.send(data)
    else:
        print(‘closing‘, conn)
        sel.unregister(conn)
        conn.close()
 
sock = socket.socket()
sock.bind((‘localhost‘, 8080))
sock.listen()
sock.setblocking(False)
sel.register(sock, selectors.EVENT_READ, accept)    #把刚生成的sock连接对象注册到select连接列表中，并交给accept函数处理
 
while True:
    events = sel.select()   #默认是阻塞，有活动连接就返回活动的连接列表
    #这里看起来是select，其实有可能会使用epoll，如果你的系统支持epoll，那么默认就是epoll
    for key, mask in events:
        callback = key.data     #去调accept函数
        callback(key.fileobj, mask) #key.fileobj就是readable中的一个socket连接对象