Event事件、进程池与线程池、协程

from threading import Event
from threading import Thread
import time
e = Event()
# e.wait()
# e.set()
def light():
    print('红灯亮了...')
    time.sleep(5)
    e.set()#set方法的作用是将wait方法的False状态改为True，
    #当e中的wait状态为False时程序会暂停，当为True状态时程序会继续运行

    print('绿灯亮了。。。')

def car(name):
    print(f'{name}正在等红灯。。。')
    e.wait()
    print(f'{name}正在加速漂移。。。')

t = Thread(target=light)
t.start()

for line in range(10):
    t1 = Thread(target=car,args=(f'{line}号car',))
    t1.start()

from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import time
#pool = ProcessPoolExecutor()#默认以CPU的个数限制进程数
#pool = ThreadPoolExecutor()默认以CPU的个数*5限制线程数

pool = ThreadPoolExecutor(5)#代表处于并发状态的最多有5个线程
#也就是在池子里面的只能有5个，但是如果池子里面有进程或线程出来了就可以有新的进程或线程进入池子。
def task():
    print('线程任务开始了。。。')
    time.sleep(1)
    print('线程任务结束了。。。')

for line in range(5):#这里对线程的提交是异步提交
    pool.submit(task)
    
#打印结果：
线程任务开始了。。。
线程任务开始了。。。
线程任务开始了。。。
线程任务开始了。。。
线程任务开始了。。。
线程任务结束了。。。
线程任务结束了。。。
线程任务结束了。。。
线程任务结束了。。。
线程任务结束了。。。

from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import time
#pool = ProcessPoolExecutor()#默认以CPU的个数限制进程数
#pool = ThreadPoolExecutor()默认以CPU的个数*5限制线程数

pool = ThreadPoolExecutor(5)#代表处于并发状态的最多有5个线程
#也就是在池子里面的只能有5个，但是如果池子里面有进程或线程出来了就可以有新的进程或线程进入池子。
def task(res):
    print(res)
    print('线程任务开始了。。。')
    time.sleep(1)
    print('线程任务结束了。。。')
    return 123

def call_back(res):
    print(type(res))
    res2 = res.result()#接收到的是task return的值
    print(res2)
# 异步提交任务
# pool.submit('传函数地址').add_done_callback('回调函数地址')

for line in range(5):
    pool.submit(task,1).add_done_callback(call_back)
    #这里的1是传入task的参数

pool.shutdown()#让线程池内的所有任务执行完成后，才执行下面的代码
print('所有线程运行结束')

#执行结果
1
线程任务开始了。。。
1
线程任务开始了。。。
1
线程任务开始了。。。
1
线程任务开始了。。。
1
线程任务开始了。。。
线程任务结束了。。。
线程任务结束了。。。
线程任务结束了。。。
线程任务结束了。。。
<class 'concurrent.futures._base.Future'>
<class 'concurrent.futures._base.Future'>
<class 'concurrent.futures._base.Future'>
<class 'concurrent.futures._base.Future'>
123
123
线程任务结束了。。。
123
123
<class 'concurrent.futures._base.Future'>
123
所有线程运行结束

from concurrent.futures import ThreadPoolExecutor
import requests
import re
import uuid

pool = ThreadPoolExecutor(100)

#1.发送请求函数
def get_page(url):
    response = requests.get(url)
    print("获取电影信息标记",response)
    return response

#2.解析主页获取视频ID号
def parse_index(response):
    id_list = re.findall('<a href="video_(.*?)".*?>',
                         response.text,re.S)
    return id_list

#3.解析视频详情页获取真实视频链接
def parse_detail(res):
    response = res.result()
    movie_detail_url = re.findall('srcUrl="(.*?)"',response.text,re.S)[0]
    print(f'往视频链接：{movie_detail_url}发送请求。。。')
    #异步向视频详情页发送请求，
    print("电影链接标记",movie_detail_url)
    pool.submit(get_page,movie_detail_url).add_done_callback(save_movie)
    return movie_detail_url
#保存视频
def save_movie(res):
    print('biaoji')
    movie_response = res.result()
    print('biaoji',movie_response)
    name = str(uuid.uuid4())
    print(f'{name}视频开始保存。。。')
    with open(f'{name}.mp4','wb') as f:
        f.write(movie_response.content)
    print('视频下载完毕')

if __name__ == '__main__':
    #1.访问视频主页获取数据
    index_response = get_page('https://www.pearvideo.com/')
    #2.解析主页获取所有视频ID号
    id_list = parse_index(index_response)
    #3.循环对每个视频详情页链接进行拼接
    for id in id_list:
        print(id)
        detail_url = 'https://www.pearvideo.com/video_' + id
        #异步提交爬取视频详情页，把返回的数据，交给parse_detail(回调函数)
        pool.submit(get_page,detail_url).add_done_callback(parse_detail)

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

def play(name):
    print('%s play 1'%name)
    gevent.sleep(1)
    print('%s play 2'%name)
#创建一个协程对象g1，spawn括号内第一个参数是函数名，如eat，
# 后面可以有多个参数，可以是位置实参或关键字实参，
# 都是传给函数eat的
g1 = gevent.spawn(eat,'egon')
g2 = gevent.spawn(play,name='egon')
g1.join()
g2.join()
print('主')

egon eat 1
egon play 1
egon play 2
egon eat 2
主

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


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

def play(name):
    print('%s play 1'%name)
    time.sleep(1)
    print('%s play 2'%name)

g1 = gevent.spawn(eat,'egon')
g2 = gevent.spawn(play,name='egon')
g1.join()
g2.join()
print('主')

egon eat 1
egon play 1
egon play 2
egon eat 2
主

#服务端
from gevent import monkey;monkey.patch_all()
import gevent
import socket
import threading
#如果不想用money.patch_all()打补丁,可以用gevent自带的socket
# from gevent import socket
# s=socket.socket()

def server(server_ip,port):
    s = socket.socket()
    s.bind((server_ip,port))
    s.listen(5)
    while True:
        print(threading.current_thread().name)
        conn,addr = s.accept()
        gevent.spawn(talk,conn,addr)

def talk(conn,addr):
    try:
        while True:
            res = conn.recv(1024).decode('utf-8')
            print(f'client{addr[0]}:{addr[1]} msg:{res}')
            conn.send(res.upper().encode('utf-8'))
    except Exception as e:
        print(e)
    finally:
        conn.close()

if __name__ == '__main__':
    server('127.0.0.1',6666)

#客户端
import threading
import socket


def client(server_ip,port):
    c = socket.socket()
    #套接字对象一定要加到函数内，即局部名称空间内，
    # 放在函数外则被所有线程共享，则大家公用一个套接字对象，
    # 那么客户端端口永远一样了
    c.connect((server_ip,port))
    count = 0
    while True:
        c.send(f'{threading.current_thread().getName()} say hello {count}'.encode('utf-8'))
        msg = c.recv(1024)
        print(msg.decode('utf-8'))
        count += 1

if __name__ == '__main__':
    for i in range(10):
        t = threading.Thread(target=client,args=('127.0.0.1',6666))
        t.start()