【0919 | Day 35】线程queue/线程池和进程池/协程

目录

• socket多线程
  • 客户端
  • 服务端
• 线程queue
  • 用法一：先进先出(Queue)
  • 用法二：先进后出(LifoQueue)
  • 用法三：优先级(PriorityQueue)
• 线程定时器
  • 介绍
• 进程池和线程池
  • 介绍
  • 例子(进程池)
  • 例子(线程池)
• 同步和异步
  • 介绍
• 协程
  • 介绍
    • 什么是协程？
    • 为什么要有协程？
    • 什么样的协程有意义？
    • 协程的优点是什么？
    • 携程的缺点是什么？
  • 例子
    • 情况一：单纯地切换反而会降低运行效率
    • 情况二：yield不能实现io切换
• gevent模块
  • 介绍
  • 例子
    • 情况一：gevent
    • 情况二：from gevent import money;money.patch_all()

socket多线程

客户端

import socket
from threading import Thread

def client_demo():
    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    
    client.connect(('192.168.11.199', 8010))
    while True:
        msg = f'{currentThread().name}'
        if len(msg) == 0: break
        client.send(msg.encode('utf-8'))
        feedback = client.recv(1024)
        print(feedback.decode('utf-8'))
     
    client.close()
    
if __name__ == '__main__':
    for i in range(5):
        t = Thread(target=client_demo)
        t.start()
            

服务端

import socket
from threading import Thread

def talk(conn):
    while True:
        try:
            msg = conn.recv(1024)
            if len(msg) == 0: break
            conn.send(msg.upper())
        except connectionResetError
        	print('客户端关闭了一个链接')
            break
    conn.close()
    
    
def serve_demo():
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.bind(('192.168.11.199', 8010))
    server.listen(5)
    
    while True:
        conn, addr = server.accept()
        t = Thread(target=talk, args(conn,))
        t.start()
        
if __name__ == '__main__':
    server_demo()

线程queue

用法一：先进先出(Queue)

import queue

q = queue.Queue()
q.put('123')
q.put('qweqwe')
print(q.get())
print(q.get())
q.task_done()
q.task_done()

123
qweqwe

用法二：先进后出(LifoQueue)

堆栈

import queue

q = queue.LifoQueue() 
q.put('粉红色的背心儿')
q.put('粉红色的裤子')
q.put('欧文的各种设备')
print(q.get())
print(q.get())
print(q.get())

欧文的各种设备
粉红色的裤子
粉红色的背心儿

用法三：优先级(PriorityQueue)

通常这个元组的第一个值是int类型

import queue

# 可以根据优先级取数据
q = queue.PriorityQueue() 
q.put((50, '吴磊'))
q.put((80, '陈飞宇'))
q.put((1, '欧阳娜娜'))
print(q.get())
print(q.get())
print(q.get())

(1, '欧阳娜娜')
(50, '吴磊')
(80, '陈飞宇')

线程定时器

介绍

启动某个线程的倒计时

from threading import Thread
import time

def task():
    print('线程执行！')
    time.sleep(1)
    print('线程结束！')
    

t = Timer(3, task) #3秒后执行task
t.start()

==》时间过去3秒

线程执行了

==》时间过去2秒

线程结束了

进程池和线程池

介绍

• 池：限制进程数或线程数
• 限制时间：当并发的任务数量远远大于计算机所能承受的范围，即无法一次性开启过多的任务数量

例子(进程池)

from concurrent.future import ProcessPoolExecutor, ThreadPoolExecutor
from threading import currentThread
from multiprocessing import current_process
import time

def task(i):
    print(f'{current_process().name}在执行任务{i}')
    time.sleep(1)
    return i**2

if __name__ == '__main__':
    pool = ProcessPoolExecutor(3) # 池子里只有3个进程
    fu_list = []
    for i in range(9):
        future = pool.submit(task, i) # task任务要做9次, 3个进程负责做这个事
        fu_list.append(future)
    pool.shutdown() # 关闭了池的入口,会等待所有的任务执行完,结束阻塞
    for fu in fu_list:
        print(fu.result())

进程 SpawnProcess-1 在执行任务 0
进程 SpawnProcess-2 在执行任务 1
进程 SpawnProcess-3 在执行任务 2

进程 SpawnProcess-1 在执行任务 3
进程 SpawnProcess-2 在执行任务 4
进程 SpawnProcess-3 在执行任务 5

进程 SpawnProcess-1 在执行任务 6
进程 SpawnProcess-2 在执行任务 7
进程 SpawnProcess-3 在执行任务 8

0
1
4
9
16
25
36
49
64

例子(线程池)

from concurrent.future import ProcessPoolExecutor, ThreadPoolExecutor
from threading import currentThread
from multiprocessing import current_process
import time

def task(i):
    print(f'{currentThread().name}在执行任务{i}')
    time.sleep(1)
    return i**2

if __name__ == '__main__':
    pool = ThreadPoolExecutor(3) # 池子里只有3个线程
    fu_list = []
    for i in range(9):
        future = pool.submit(task, i) # task任务要做9次, 3个线程负责做这个事
        fu_list.append(future)
    pool.shutdown() # 关闭了池的入口,会等待所有的任务执行完,结束阻塞
    for fu in fu_list:
        print(fu.result())

ThreadPoolExecutor-0_0 在执行任务 0
ThreadPoolExecutor-0_1 在执行任务 1
ThreadPoolExecutor-0_2 在执行任务 2

ThreadPoolExecutor-0_2 在执行任务 3
ThreadPoolExecutor-0_0 在执行任务 4
ThreadPoolExecutor-0_1 在执行任务 5

ThreadPoolExecutor-0_0 在执行任务 6
ThreadPoolExecutor-0_1 在执行任务 7
ThreadPoolExecutor-0_2 在执行任务 8

0
1
4
9
16
25
36
49
64

同步和异步

介绍

• 同步：提交了一个任务，必须等任务执行完拿到返回值，才能执行下一行代码（一般不使用）
• 异步：提交了一个任务，不等执行完，就可以直接执行下一行代码

协程

介绍

什么是协程？

• 单线程下实现并发

为什么要有协程？

• 协程本质是程序员抽象出来的概念，操作系统根本不知道协程的存在

  • 操作系统 == 佛祖

  • 线程 == 孙悟空

  • 协程 == 猴毛

• 通俗的说，孙悟空的一切活动都在佛祖的股掌之中。一旦某个小猴子遇到io或者磨磨蹭蹭耽误太长时间了，就需要交出自己的CPU执行权限（金箍棒），由别的猴子（线程）去做

• 可是这样，孙悟空觉得不服气，凭什么？？？我也不想等那么长时间的io啊！因此，孙悟空一气之下，拔了一撮脑袋上的猴毛（协程），这些腾空出世的小猴子由他调遣，所以这下佛祖就没辙啦！人一多，办事效率就高了嘛

什么样的协程有意义？

• 只有遇到io切换才有意义

协程的优点是什么？

• 应用程序控制切换要比操作系统切换快得多（或者说协程的切换开销更小，属于程序级别的切换，操作系统完全感知不到，更加轻量级）
• 单线程内就可以实现并发的效果，最大限度地利用CPU

携程的缺点是什么？

• 协程的本质是在单线程下，无法利用多核，可以是一个程序开启多个进程，每个进程开启多个线程，每个线程开启协程
• 协程指的是单个线程，因而一旦协程出现阻塞，将会阻塞整个线程

例子

情况一：单纯地切换反而会降低运行效率

import time

def consumer(res):
    '''任务1:接收数据，处理数据'''
    pass

def producer():
    '''任务2:生产数据'''
    res = []
    for i in range(10000000):
        res.append(i)
    return res

start = time.time()

#串行执行
res = producer()
consumer(res) #写成consumer(producer())会降低执行效率
stop = time.time()
print(stoop - start)


#基于yield并发执行
import time 
def consumer():
    '''任务1:接收数据，处理数据'''
    while True:
        x = yield

def producer():
    '''任务2:生产数据'''
    g = consumer()
    next(g)
    for i in range(10000000)
        g.send(i)
        
start = time.time()

#基于yield保存状态，实现两个任务直接来回切换，即并发的效果
#PS:如果每个任务中都加上打印，那么明显地看到两个任务的打印是你一次我一次，即并发执行的

producer()

stop = time.time()
print(stop - start)

情况二：yield不能实现io切换

import time 

def consumer():
    '''任务1：接收数据, 处理数据'''
    while True:
        x = yield
        
def producer():
    '''任务2：生产数据'''
    g = consumer()
    next(g)
    for i in range(10000000):
        g.send(i)
        time.sleep(2)
        
start = time.time()
producer() #并发执行，但是任务producer遇到io就会阻塞住，并不会切到该线程内的其他任务去执行

stop = time.time()
print(stop - start)

gevent模块

介绍

• 实现并发同步或异步编程

例子

情况一：gevent

遇到io阻塞时会自动切换任务

import gevent

def eat(name):
    print(f'{name} eat 1')
    gevent.sleep(2)
    print(f'{name} eat 2')

def play(name):
    print(f'{name} play 1')
    gevent.sleep(3)
    print(f'{name} play 2')


g1 = gevent.spawn(eat, 'ada')
g2 = gevent.spawn(play, name = 'ada')
g1.join()  #等待g1结束
g2.join()  #等待g2结束
print('主')

情况二：from gevent import money;money.patch_all()

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

def eat():
    print('eat 1')
    time.sleep(2)
    print('eat 2')

def play():
    print('play 1')
    # 疯狂的计算没有io
    time.sleep(3)
    print('play 2')

start = time.time()
g1 = gevent.spawn(eat)
g2 = gevent.spawn(play)
g1.join()  #等待g1结束
g2.join()  #等待g2结束
end = time.time()
print(end-start)