day 31 小结

目录

• Event事件
• 进程池与线程池
• 协程
• gevent实现协程
• TCP服务端实现协程
• IO模型

Event事件

​ 作用: 用来控制线程的执行

​ 由一些线程去控制另一些线程

from threading import Event
from threading import Thread
import time
# 调用Event类实例化一个对象
e = Event()

# 若该方法出现在任务中,则为False,阻塞
# e.wait()  # False

# 若该方法出现在任务中,则将其他线程的Flase改为True,进入就绪态与运行态
# e.set()  # True


def light():
    print('红灯亮...')
    time.sleep(5)
    # 应该开始发送信号,告诉其他线程准备执行
    e.set()  # 将car中的False ---> True
    print('绿灯亮...')


def car(name):
    print('正在等红灯....')
    # 让所有汽车任务进入阻塞态
    e.wait()  # False
    print(f'{name}正在加速漂移....')


# 让一个light线程任务 控制多个car线程任务
t = Thread(target=light)
t.start()

for line in range(10):
    t = Thread(target=car, args=(f'童子军jason{line}号', ))
    t.start()

进程池与线程池

​ 进程池与线程池是用来控制当前程序允许创建(进程/线程)的数量

​ 作用: 保证在硬件允许的范围内创建(进程/线程)的数量

from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
import time

# ProcessPoolExecutor(5)  # 5代表只能开启5个进程
# ProcessPoolExecutor()  # 默认以CPU的个数限制进程数

pool = ThreadPoolExecutor(5)  # 5代表只能开启5个线程 -5 +1 -1 +1 -1
# ThreadPoolExecutor()  # 默认以CPU个数 * 5 限制线程数

# t = Tread()  # 异步提交
# t.start(0)



# pool.submit('传函数地址')  # 异步提交任务
# def task():
#     print('线程任务开始了...')
#     time.sleep(1)
#     print('线程任务结束了...')
#
#
# for line in range(5):
#     pool.submit(task)


# 异步提交任务
# pool.submit('传函数地址').add_done_callback('回调函数地址')
def task(res):
    # res == 1
    print('线程任务开始了...')
    time.sleep(1)
    print('线程任务结束了...')
    return 123


# 回调函数
def call_back(res):
    print(type(res))
    # 注意: 赋值操作不要与接收的res同名
    res2 = res.result()
    print(res2)


for line in range(5):
    pool.submit(task, 1).add_done_callback(call_back)


# 会让所有线程池的任务结束后,才往下执行代码
# pool.shutdown()

print('hello')

协程

- 进程:	资源单位
- 线程:    执行单位
- 协程:    在单线程下实现并发

注意: 协程不是操作系统资源,他是程序起的名字,为让单线程能实现并发

协程的目的:

• 操作系统
  多道技术, 切换+ 保存状态

  - 遇到IO
  - CPU执行时间过长
  

• 协程: 通过手动模拟操作系统"多道技术",实现 切换+保存状态

  • 手动实现 遇到IO切换,欺骗操作系统误以为没有IO操作
    • 单线程,遇到IO, 切换+保存状态
    • 单线程 计算密集型,来回切换+保存状态,反而效率更低

  优点: 在IO密集型的情况下,会提高效率

  缺点: 若在计算密集型的情况下,来回切换,反而效率更低

• 如何实现协程: 切换+保存状态

  • yield: 保存状态

  import time
  
  def func1():
      while True:
          10000000+1
          yield
  
  def func2():
      
      g = func1() # 启动生成器 ，g 为生成器对象
      for i in range(10000000):
          time.sleep(10)  # 模拟IO，yield并不会捕捉到并自动切换
          i+1
          next(g)
  
  start = time.time()
  func2()
  stop = time.time()
  

gevent实现协程

导入 第三方库: from gevent import monkey,spawn,joinall

monkey.patch_all() 可以监听程序下所有的IO操作

spawn和joinall可以实现保存状态+切换

from gevent import monkey
monkey.patch_all()  # 可以监听该程序下所有的IO操作
import time
from gevent import spawn,joinall  # 用于做切换+保存状态

def func1():
    print('1')
    # IO操作
    time.sleep(1)
    
def func2():
    print('2')
    time.sleep(1)
    
def func3():
    print('3')
    time.sleep(1)
    
    
start_time = time.time()

s1 = spawn(func1)  # 传任务
s2 = spawn(func2)
s3 = spawn(func3)

# s2.join()  # 发送信号,相当于等待自己 (在单线程的情况下)
# s1.join()     
# s3.join()

# 必须传序列类型，如列表和元组
joinall([s1, s2, s3])

end_time = time.time()

print(end_time - start_time)

TCP服务端实现协程

# 服务端
from gevent import monkey
monkey.patch_all()  # 检测IO

import socket
import time
from threading import Thread
from gevent import spawn

server = socket.socket()

server.bind(
    ('127.0.0.1', 9527)
)

server.listen(5)
print('启动服务端...')


# 线程任务,执行接收客户端消息与发送消息给客户端
def working(conn):
    while True:
        try:
            data = conn.recv(1024).decode('utf-8')
            if len(data) == 0:
                break
            print(data)
            # time.sleep(1)
            send_data = data.upper().encode('utf-8')
            conn.send(send_data)

        except Exception as e:
            print(e)
            break

    conn.close()


def server2():
    while True:
        conn, addr = server.accept()
        # print(addr)
        # t = Thread(target=working, args=(conn,))
        # t.start()
        spawn(working, conn)


if __name__ == '__main__':
    s1 = spawn(server2)
    s1.join()

# 客户端
import socket
import time
from threading import Thread, current_thread


def client():
    client = socket.socket()
    client.connect(
        ('127.0.0.1', 9527)
    )
    print('启动客户端...')
    number = 0
    while True:
        send_data = f'{current_thread().name} {number}'
        client.send(send_data.encode('utf-8'))
        data = client.recv(1024)
        print(data.decode('utf-8'))
        number += 1


# 模拟了300个用户并发去访问服务端
for i in range(300):
    t = Thread(target=client)
    t.start()

IO模型

1. 阻塞I/O模型

2. 非阻塞I/O模型

3. 多路复用I/O模型

4. 信号驱动I/O模型

5. 异步I/O模型