【python自动化第十一篇】

时间:2023-03-08 15:49:02
【python自动化第十一篇】

【python自动化第十一篇:】

课程简介

  • gevent协程
  • select/poll/epoll/异步IO/事件驱动
  • RabbitMQ队列

上节课回顾

  • 进程:

    • 进程的诞生时为了处理多任务,资源的隔离,提供程序运行的所有数据
    • 进程就是一个程序运行所需要的资源集合
    • 每个进程的数据是独立的
    • 每个进程至少有一个线程
    • 适用于CPU密集型程序(金融分析等。。)
  • 线程:

    • 线程数据是共享的
    • 线程依赖于进程运行
    • 适用于IO密集型程序(socket,web,爬虫)
  • 总结:

    • 一个进程的多个线程可以充分利用多和cpu
    • 进程间数据共享的代价是高昂的,所以要尽量避免进程间的数据共享
    • 线程要修改同一份数据,必须要加锁(互斥锁mutex)
    • event :线程间交互
  • 多进程

    • multiprocessing
      • pipe
      • queue :FIFO,LIFO,priority
      • pipe和queue实现的是进程间的数据传递和通信
      • manager实现了多进程间的数据共享
  • 生产者消费者模型

    • 解耦(降低程序之间的关联)
    • 提高程序的运行效率

协程

协程:又称微线程,是一种用户态的轻量级线程

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

2 . 协程能够保存上一次调用时的状态(所有局部状态的一个特定组合),每一次过程重入时,就相当于进入上一次调用的状态,换一种说法就是进入上一次离开时所处的逻辑流的位置

  • 协程的好处:
  1. 无需上下文切换的开销
  2. 无需原子操作锁定以及同步开销
  3. 方便切换控制流,简化编程模型
  4. 高并发+高拓展性+低成本:一个CPU支持上万个协程都不是问题,所以很适合高并发处理
  • 协程的缺点:
  1. 无法利用多核资源
  2. 进行阻塞操作(如IO)会阻塞整个程序
  • 小例子:
import queue
import time def consumer(name):
"""
定义消费者
:param name:
:return:
"""
print("-->starting eating baozi!!")
while True:
new_baozi = yield
print("[%s] is eating baozi %s"%(name,new_baozi))
time.sleep(1) #程序中间有停顿,则会导致程序串行执行
def producer():
"""
定义生产者
:return:
"""
r = con.__next__()
r = con2.__next__()
n = 0
while n<5000000000: #循环5000个包子
n+=1
con.send(n)
con2.send(n)
# print("\033[32;1m[producer]\033[0m is making baozi %s"%n)
if __name__ == '__main__':
con = consumer("c1")
con2 = consumer("c2")
p = producer()

总结:

协程从上述程序中得出的定义:

  1. 必须在只有一个单线程里实现并发
  2. 修改共享数据不需要加锁
  3. 用户程序里自己保存多个控制流的上下文栈
  4. 一个协程遇到其他的IO操作就自动切换到其他协程
  • 通过greenlet模块实现的协程
from greenlet import greenlet
import time def test1():
print(12)
time.sleep(2) #等待之后再切换
gr2.switch() #切换到gr2
print(34)
gr2.switch()
def test2():
print(56)
gr1.switch()
print(78)
gr1 = greenlet(test1)
gr2 = greenlet(test2) gr1.switch()

不用变成生成器,可以在任意位置切换,但是不能够实现 一个协程遇到其他的IO操作就自动切换到其他协程

  • gevent实现并发同步或者异步编程
import gevent

def func1():
print('\033[31;1mfunc1开始运行\033[0m')
gevent.sleep(2)
print("\033[31;1mfunc1继续运行\033[0m") def func2():
print('\033[32;1mfunc2开始运行\033[0m')
gevent.sleep(1)
print("\033[32;1mfunc2继续运行\033[0m")
def func3():
print("333333")
gevent.sleep(1)
print("44444")
gevent.joinall([
gevent.spawn(func1),
gevent.spawn(func2),
gevent.spawn(func3)
])

输出结果如下:

func1开始运行

func2开始运行

333333

func2继续运行

44444

func1继续运行

  • 同步和异步的性能差异对比
import gevent

def task(pid):
gevent.sleep(0.5)
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)] #列表生成式
threads = []
for i in range(10):
threads.append(gevent.spawn(task,i))
gevent.joinall(threads)
print("synchonous")
synchronous() #串行(同步)
# print("asynchonoous..")
# asynchronous() #并行(异步)
  • 并发爬虫实例
import gevent
from gevent import monkey
monkey.patch_all() #打一发补丁
from urllib.request import urlopen
import time def pa_web_page(url):
print("starting get url ",url)
req = urlopen(url) #打开链接
data = req.read() #读取数据
print(data)
print("%d bytes recieved from %s"%(len(data),url))
# start_time = time.time()
# pa_web_page("http://www.xiaohuar.com")
# pa_web_page("http://www.autohome.com.cn/beijing/")
# print("cost time is %s"%(time.time()-start_time)) start_time2 = time.time()
gevent.joinall([
gevent.spawn(pa_web_page,"http://www.xiaohuar.com"),
gevent.spawn(pa_web_page,"http://www.autohome.com.cn/beijing/"),
gevent.spawn(pa_web_page,"http://www.sina.com"),
])
print("cost time is %s"%(time.time()-start_time2))
  • 基于socket的并发连接测试

    (1).socket_server 代码如下:
import socket
import gevent
from gevent import monkey
monkey.patch_all()
def server(port):
s = socket.socket()
s.bind(("0.0.0.0",port))
s.listen(500)
while True:
client,addr=s.accept()
gevent.spawn(handle_request,client)
def handle_request(conn):
try:
while True:
data = conn.recv(1024)
print("recv",data)
conn.send(data)
if not data:
conn.shutdown(socket.SHUT_WR)
except Exception as e:
print(e)
finally:
conn.close()
if __name__ == '__main__':
server(8888)

(2). socket_client for threading 代码如下:

import socket
import threading
def sock_conn():
client = socket.socket()
client.connect(("localhost",8888))
count = 0
while True:
client.send(("hello %s"%count).encode("utf-8"))
data = client.recv(1024)
print("[%s] recv server %s"%(threading.get_ident(),data.decode()))
count+=1
client.close() for i in range(100): #测试并发连接个数
t = threading.Thread(target=sock_conn)
t.start()
  • 此时会有一个弊端就是当并发高于1024就会不稳定

    (3). socket_client normal 代码如下:
import socket
HOST = "localhost"
PORT = 8888
c = socket.socket()
c.connect((HOST,PORT))
while True:
msg = bytes(input(">>"),encoding="utf-8")
c.sendall(msg)
data = c.recv(1024)
print("Recived",repr(data))

事件驱动和异步IO

  • 通常我们写服务器处理模型的程序时,遇到的模型:

    (1)每收到一个请求,创建一个新的进程,来处理该请求;

    (2)每收到一个请求,创建一个新的线程,来处理该请求;

    (3)每收到一个请求,放入一个事件列表,让主进程通过非阻塞I/O方式来处理请求

    特点:

    第(1)中方法,由于创建新的进程的开销比较大,所以,会导致服务器性能比较差,但实现比较简单。

    第(2)种方式,由于要涉及到线程的同步,有可能会面临死锁等问题。

    第(3)种方式,在写应用程序代码时,逻辑比前面两种都复杂。

    综合考虑各方面因素,一般普遍认为第(3)种方式是大多数网络服务器采用的方式

  • 事件驱动类型介绍:

    在UI编程中,常常要对鼠标点击进行相应,首先如何获得鼠标点击呢?

    方式一:创建一个线程,该线程一直循环检测是否有鼠标点击,那么这个方式有以下几个缺点:

  1. CPU资源浪费,可能鼠标点击的频率非常小,但是扫描线程还是会一直循环检测,这会造成很多的CPU资源浪费;如果扫描鼠标点击的接口是阻塞的呢?

  2. 如果是堵塞的,又会出现下面这样的问题,如果我们不但要扫描鼠标点击,还要扫描键盘是否按下,由于扫描鼠标时被堵塞了,那么可能永远不会去扫描键盘;

  3. 如果一个循环需要扫描的设备非常多,这又会引来响应时间的问题;

    所以,该方式是非常不好的。

    方式二:就是事件驱动模型

    目前大部分的UI编程都是事件驱动模型,如很多UI平台都会提供onClick()事件,这个事件就代表鼠标按下事件。事件驱动模型大体思路如下:

  4. 有一个事件(消息)队列;

  5. 鼠标按下时,往这个队列中增加一个点击事件(消息);

  6. 有个循环,不断从队列取出事件,根据不同的事件,调用不同的函数,如onClick()、onKeyDown()等;

  7. 事件(消息)一般都各自保存各自的处理函数指针,这样,每个消息都有独立的处理函数;

echo 'Hello world!';