协程

直接使用线程或者进程发起异步

from multiprocessing import Process
from threading import Thread

import time


def task():
    time.sleep(2)
    print('run...')


# 线程
# p = Thread(target=task)
# p.start()
# print('over')
# 进程
if __name__ == '__main__':
    p = Process(target=task)
    p.start()
    print('over')

获取异步任务结果的方式

　　爬虫：

　　　　1.获取到HTML文档

　　　　2.从文档中取出需要的数据

　　回调：其实说的是回调函数

　　　　　给异步任务绑定一个函数，当任务完成时会自动调用该函数

　　具体使用：

　　　　当你往pool中添加了一个异步任务，会返回一个表示结果的对象

　　　　有一个对象绑定方法 add_done_callback 需要一个函数做为参数

　　　　注意：回调函数 必须有且只有一个参数 就是对象本身

　　　　　　通过对象点(.)result来获取结果

　　优点：不用原地等待，任务结果可以立即获取到

import requests
from concurrent.futures import ThreadPoolExecutor
import threading


def get_data(url):
    res = requests.get(url)
    # print(res.text)
    print('%s正在处理%s' % (threading.current_thread().name, url))
    # 方式3 直接调用处理函数 生产者和消费者 强行耦合在一起
    # parser_data(res.text)
    return res.text, url


def parser_data(f):
    # print(f)
    # res = f.result()
    print('解析长度为:%s，地址为：%s' % (f.result()[0], f.result()[1]))


urls = ['https://www.baidu.com/', 'https://www.bilibili.com/', 'http://hs.blizzard.cn/landing']
pool = ThreadPoolExecutor()
fs = []
for url in urls:
    f = pool.submit(get_data, url)  # 提交任务
    f.add_done_callback(parser_data)
    # 方式一 把并发变成了串行
    # data = f.result()
    # parser_data(data)
    # fs.append(f)
# 方式2必须等待全部完成才返回结果
# pool.shutdown()
# for f in fs:
#     data = f.result()
#     parser_data(data[0], data[1])

如何为线程或进程绑定回调函数

from threading import Thread
from concurrent.futures import ThreadPoolExecutor
import os


#
#
# def task(callback):
#     print('run ...')
#     callback()
#
#
# def callback():
#     print('这是回调函数。。。')
#
#
# t = Thread(target=task, args=(callback,))
# t.start()
def task():
    print('task...run')


def call_back(obj):
    print(os.getpid())
    print(obj.result())


print('主：', os.getpid())
pool = ThreadPoolExecutor()
f = pool.submit(task)
f.add_done_callback(call_back)

from multiprocessing import Process
from concurrent.futures import ProcessPoolExecutor
import os


# def task(callback):
#     print('run ...')
#     callback()
#
#
# def callback():
#     print('这是回调函数。。。')
#
#
# if __name__ == '__main__':
#     t = Process(target=task, args=(callback,))
#     t.start()


def task():
    print('task...run')


def call_back(obj):
    print(os.getpid())
    print(obj.result())


if __name__ == '__main__':
    print('主：', os.getpid())
    pool = ProcessPoolExecutor()
    f = pool.submit(task)
    f.add_done_callback(call_back)

线程队列

queue该模块下提供了一些常见的数据容器但是它们仅仅是容器 每一个都有数据共享这个特点

# from  multiprocessing import Queue
from queue import Queue, LifoQueue, PriorityQueue

# q=Queue()
# q.put(1)
# q.put(2)
# print(q.get())
# print(q.get())
# print(q.get(timeout=2))

# 后进先出队列(堆栈)
# q = LifoQueue()
# q.put(1)
# q.put(2)
# print(q.get())

# 优先级队列
# 需要传入一个元组类型 第一个是优先级 第二个是值
q = PriorityQueue()
# q.put((-111111, "abc"))
# q.put((2, "abc"))
q.put(("a", "hello world1"))
q.put(("A", "hello world2"))
q.put(("c", "hello world"))

print(q.get())

事件 美国911事件 圆明园事件 

　　事件是一个通知信息，表示什么时间发生了什么事情

　　用于线程间通讯

　　线程间本来就是数据共享的 也就是说 即使没有事件这个东西 也是没问题的

　　线程之间，执行流程是完全独立的，一些时候可能需要知道另一个线程发生了什么 然后采取一些行动

　　这个时候就可以使用事件来简化代码

　　事件其实就是帮你维护了一个bool值

　　在bool为True之前 wait函数将一直阻塞，这样一来就避免了不断的询问对方的状态

　　假设 有两条线程 一个用于开启服务器 一个用于连接服务器

　　连接服务器一定要保证服务器已经启动成功了，服务器启动需要花费一些时间

import time, random

from threading import Thread, Event

# 使用事件
# 一个事件
boot = Event()


def boot_server():
    print('正在启动服务器。。。')
    time.sleep(random.randint(2, 5))
    print('服务器启动成功！')
    boot.set()


def connect_server():
    print('开始尝试连接。。。')
    boot.wait()  # 是一个阻塞函数 会一直等到set()函数被调用
    print('服务器连接成功！')

# 不使用事件
# 默认未启动
# is_boot = False
# 
# 
# def boot_server():
#     global is_boot
#     print('正在启动服务器。。。')
#     time.sleep(random.randint(2, 5))
#     print('服务器启动成功！')
#     #   修改状态为True
#     is_boot = True
# 
# 
# def connect_server():
#     while True:
#         if is_boot:
#             print('连接服务器成功！')
#             break
#         else:
#             print('连接失败 服务器未启动！')
#             time.sleep(1)
# 
# 
# t1 = Thread(target=boot_server)
# t1.start()
# t2 = Thread(target=connect_server)
# t2.start()

协程

　　协程就是要用单线程实现并发

　　GIL导致多个线程不能并行，效率低

　　CPython中多个线程不能并行

　　既然多个线程也无法提高执行效率 还有没有必须要开线程

　　例子：只有一个岗位 但是有十个任务，请十个人也提高不了效率，反而增加了系统开销

　　现在只有一个线程，那要如何并发的处理十个任务

　　一个线程如何能并发

　　　　多道技术

　　　　切换+保存状态

　　　　首先任务其实就是一堆代码，一堆代码可以组成一个函数

　　　　如何能使得可以在多个函数之间进行切换

　　协程指的就是一个线程完成并发执行

　　在CPyrhon中 如果你的任务是计算密集型 使用协程是无法提高效率的 反而因为切换任务导致效率降低

　　　　只能靠进程了

　　IO密集型 多线程 会比多进程效率高 因为线程的开销比进程小很多

　　本质上协程还是只有一个线程 所以一旦遇到IO操作 整个线程就卡住了

　　协程仅在以下场景能够提高效率

　　　　1.任务是IO密集型

　　　　2.一定要可以检测IO操作 并且在IO即将阻塞时 切换到计算任务 从而使得CPU尽可能 多得执行你的线程

import time


# def task1():
#     for i in range(100000):
#         time.sleep(10000)
#         yield
#
#
# def task2():
#     g = task1()
#     for i in range(100000):
#         next(g)
#
#
# start_time = time.time()
# task2()
# print(time.time() - start_time)

def task1():
    for i in range(100000):
        1 + 1


def task2():
    for i in range(100000):
        1 + 1


start_time = time.time()
task1()
task2()
print(time.time() - start_time)

greenlet　　
greenlet是对yield进行
　　简化了书写但是它不能检测IO操作

import greenlet

import time


def task1():
    print('task1 run...')
    time.sleep(20)
    g2.switch()
    print('task1 over...')


def task2():
    print('task2 run...')
    print('task2 over...')
    g1.switch()


g1 = greenlet.greenlet(task1)
g2 = greenlet.greenlet(task2)
g1.switch()

greenlet

　　无法检测IO操作

　　gevent既可单线程实现并发 又可以检测IO操作

import gevent
import gevent.monkey

gevent.monkey.patch_all()
# 打补丁的代码必须放在导入模块之前
import time
import threading


def task1():
    # for i in range(1000):
    print('task1')
    print(threading.current_thread())
    time.sleep(1)


def task2():
    # for i in range(1000):
    print(task2)
    print(threading.current_thread())


g1 = gevent.spawn(task1)
g2 = gevent.spawn(task2)

# 协程创建完成后 必须调用join函数 否则任务不会开启 可以理解为线程中的start函数
# g1.start()
# g2.start()
# g1.join()
# g2.join()

gevent.joinall([g1, g2])
print('over')