#!/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连接对象