Python并发编程(线程队列,协程,Greenlet,Gevent)

时间  2019-11-17
标签 python 并发 编程 线程 队列 greenlet gevent 栏目 Python 繁體版
原文   原文链接

线程队列

　　线程之间的通讯咱们列表行不行呢，固然行，那么队列和列表有什么区别呢？python

　　queue队列 ：使用import queue，用法与进程Queue同样nginx

　　queue is especially useful in threaded programming when information must be exchanged safely between multiple threads.git

　　class queue.Queue(maxsize=0) #先进先出 
import queue #不须要经过threading模块里面导入，直接import queue就能够了，这是python自带的
#用法基本和咱们进程multiprocess中的queue是同样的
q=queue.Queue()
q.put('first')
q.put('second')
q.put('third')
# q.put_nowait() #没有数据就报错，能够经过try来搞
print(q.get())
print(q.get())
print(q.get())
# q.get_nowait() #没有数据就报错，能够经过try来搞
'''
结果(先进先出):
first
second
third
'''

先进先出示例代码

　　class queue.LifoQueue(maxsize=0) #last in fisrt outgithub

import queue

q=queue.LifoQueue() #队列，相似于栈，栈咱们提过吗，是否是先进后出的顺序啊
q.put('first')
q.put('second')
q.put('third')
# q.put_nowait()

print(q.get())
print(q.get())
print(q.get())
# q.get_nowait()
'''
结果(后进先出):
third
second
first
'''

先进后出示例代码

　　class queue.PriorityQueue(maxsize=0) #存储数据时可设置优先级的队列编程

import queue

q=queue.PriorityQueue()
#put进入一个元组,元组的第一个元素是优先级(一般是数字,也能够是非数字之间的比较),数字越小优先级越高
q.put((-10,'a'))
q.put((-5,'a'))  #负数也能够
# q.put((20,'ws'))  #若是两个值的优先级同样，那么按照后面的值的acsii码顺序来排序,若是字符串第一个数元素相同，比较第二个元素的acsii码顺序
# q.put((20,'wd'))
# q.put((20,{'a':11})) #TypeError: unorderable types: dict() < dict() 不能是字典
# q.put((20,('w',1)))  #优先级相同的两个数据，他们后面的值必须是相同的数据类型才能比较，能够是元祖，也是经过元素的ascii码顺序来排序

q.put((20,'b'))
q.put((20,'a'))
q.put((0,'b'))
q.put((30,'c'))

print(q.get())
print(q.get())
print(q.get())
print(q.get())
print(q.get())
print(q.get())
'''
结果(数字越小优先级越高,优先级高的优先出队):
'''

优先级队列示例代码

 

这三种队列都是线程安全的，不会出现多个线程抢占同一个资源或数据的状况。数组

 

协程介绍

　　协程：是单线程下的并发，又称微线程，纤程。英文名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
#真正的协程模块就是使用greenlet完成的切换
from greenlet import greenlet

def eat(name):
    print('%s eat 1' %name)  #2
    g2.switch('taibai')   #3
    print('%s eat 2' %name) #6
    g2.switch() #7
def play(name):
    print('%s play 1' %name) #4
    g1.switch()      #5
    print('%s play 2' %name) #8

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

g1.switch('taibai')#能够在第一次switch时传入参数，之后都不须要  1

　　单纯的切换（在没有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自动切换来提高效率的问题。

　　上面这个图，是协程真正的意义，虽然没有规避固有的I/O时间，可是咱们使用这个时间来作别的事情了，通常在工做中咱们都是进程+线程+协程的方式来实现并发，以达到最好的并发效果，若是是4核的cpu，通常起5个进程，每一个进程中20个线程（5倍cpu数量），每一个线程能够起500个协程，大规模爬取页面的时候，等待网络延迟的时间的时候，咱们就能够用协程去实现并发。 并发数量 = 5 * 20 * 500 = 50000个并发，这是通常一个4cpu的机器最大的并发数。nginx在负载均衡的时候最大承载量就是5w个

　　单线程里的这20个任务的代码一般会既有计算操做又有阻塞操做，咱们彻底能够在执行任务1时遇到阻塞，就利用阻塞的时间去执行任务2。。。。如此，才能提升效率，这就用到了Gevent模块。

 

Gevent介绍

#安装
pip3 install gevent

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

#用法
g1=gevent.spawn(func,1,2,3,x=4,y=5)建立一个协程对象g1，spawn括号内第一个参数是函数名，如eat，后面能够有多个参数，能够是位置实参或关键字实参，都是传给函数eat的，spawn是异步提交任务

g2=gevent.spawn(func2)

g1.join() #等待g1结束，上面只是建立协程对象，这个join才是去执行

g2.join() #等待g2结束  有人测试的时候会发现，不写第二个join也能执行g2，是的，协程帮你切换执行了，可是你会发现，若是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('主')

遇到I/O切换

　　上例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() #必须写在最上面，这句话后面的全部阻塞所有可以识别了

import gevent  #直接导入便可
import time
def eat():
    #print()　　
    print('eat food 1')
    time.sleep(2)  #加上mokey就可以识别到time模块的sleep了
    print('eat food 2')

def play():
    print('play 1')
    time.sleep(1)  #来回切换，直到一个I/O的时间结束，这里都是咱们个gevent作得，再也不是控制不了的操做系统了。
    print('play 2')

g1=gevent.spawn(eat)
g2=gevent.spawn(play_phone)
gevent.joinall([g1,g2])
print('主')

　　咱们能够用threading.current_thread().getName()来查看每一个g1和g2，查看的结果为DummyThread-n，即假线程，虚拟线程，其实都在一个线程里面

　　进程线程的任务切换是由操做系统自行切换的，你本身不能控制

　　协程是经过本身的程序（代码）来进行切换的，本身可以控制，只有遇到协程模块可以识别的IO操做的时候，程序才会进行任务切换，实现并发效果，若是全部程序都没有IO操做，那么就基本属于串行执行了。

 

Gevent之同步与异步

from gevent import spawn,joinall,monkey;monkey.patch_all()

import time
def task(pid):
    """
    Some non-deterministic task
    """
    time.sleep(0.5)
    print('Task %s done' % pid)


def synchronous():
    for i in range(10):
        task(i)

def asynchronous():
    g_l=[spawn(task,i) for i in range(10)]
    joinall(g_l)

if __name__ == '__main__':
    print('Synchronous:')
    synchronous()

    print('Asynchronous:')
    asynchronous()
#上面程序的重要部分是将task函数封装到Greenlet内部线程的gevent.spawn。 初始化的greenlet列表存放在数组threads中，此数组被传给gevent.joinall 函数，后者阻塞当前流程，并执行全部给定的greenlet。执行流程只会在 全部greenlet执行完后才会继续向下走。

协程：同步异步对比

 

Gevent之应用举例一

from gevent import monkey;monkey.patch_all()
import gevent
import requests
import time

def get_page(url):
    print('GET: %s' %url)
    response=requests.get(url)
    if response.status_code == 200:
        print('%d bytes received from %s' %(len(response.text),url))


start_time=time.time()
gevent.joinall([
    gevent.spawn(get_page,'https://www.python.org/'),
    gevent.spawn(get_page,'https://www.yahoo.com/'),
    gevent.spawn(get_page,'https://github.com/'),
])
stop_time=time.time()
print('run time is %s' %(stop_time-start_time))

协程应用：爬虫

　　将上面的程序最后加上一段串行的代码看看效率：若是你的程序不须要过高的效率，那就不用什么并发啊协程啊之类的东西。

print('--------------------------------')
s = time.time()
requests.get('https://www.python.org/')
requests.get('https://www.yahoo.com/')
requests.get('https://github.com/')
t = time.time()
print('串行时间>>',t-s)

 

Gevent之应用举例二

　　经过gevent实现单线程下的socket并发（from gevent import monkey;monkey.patch_all()必定要放到导入socket模块以前，不然gevent没法识别socket的阻塞）

　　一个网络请求里面通过多个时间延迟time

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 socket import *

client=socket(AF_INET,SOCK_STREAM)
client.connect(('127.0.0.1',8080))


while True:
    msg=input('>>: ').strip()
    if not msg:continue

    client.send(msg.encode('utf-8'))
    msg=client.recv(1024)

客户端
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()

多线程并发多个客户端，去请求上面的服务端是没问题的

 

　　上面代码的服务端用gevent的时候为何没有用join就执行了。

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程