【python自动化第十一篇】

2023-12-23 17:07:34

【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 . 协程能够保存上一次调用时的状态（所有局部状态的一个特定组合），每一次过程重入时，就相当于进入上一次调用的状态，换一种说法就是进入上一次离开时所处的逻辑流的位置

协程的好处：

无需上下文切换的开销
无需原子操作锁定以及同步开销
方便切换控制流，简化编程模型
高并发+高拓展性+低成本：一个CPU支持上万个协程都不是问题，所以很适合高并发处理

协程的缺点：

无法利用多核资源
进行阻塞操作（如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()

总结：

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

必须在只有一个单线程里实现并发

修改共享数据不需要加锁

用户程序里自己保存多个控制流的上下文栈

一个协程遇到其他的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编程中，常常要对鼠标点击进行相应，首先如何获得鼠标点击呢？

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

CPU资源浪费，可能鼠标点击的频率非常小，但是扫描线程还是会一直循环检测，这会造成很多的CPU资源浪费；如果扫描鼠标点击的接口是阻塞的呢？
如果是堵塞的，又会出现下面这样的问题，如果我们不但要扫描鼠标点击，还要扫描键盘是否按下，由于扫描鼠标时被堵塞了，那么可能永远不会去扫描键盘；
如果一个循环需要扫描的设备非常多，这又会引来响应时间的问题；

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

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

目前大部分的UI编程都是事件驱动模型，如很多UI平台都会提供onClick()事件，这个事件就代表鼠标按下事件。事件驱动模型大体思路如下：
有一个事件（消息）队列；
鼠标按下时，往这个队列中增加一个点击事件（消息）；
有个循环，不断从队列取出事件，根据不同的事件，调用不同的函数，如onClick()、onKeyDown()等；
事件（消息）一般都各自保存各自的处理函数指针，这样，每个消息都有独立的处理函数；

echo 'Hello world!';

码农公寓

课程简介

上节课回顾

协程

事件驱动和异步IO

相关文章