当前位置： 首页 > news >正文

站长之家seo综合,给教育类做网站,网站优化seo,wordpress如何添加tdk线程间的通讯机制 消息队列 event 事件对象 当线程创建完成之后#xff0c;并不会马上执行线程#xff0c;而是等待某一事件发生#xff0c;线程才会启动 import threading# # 创建 event 对象

event threading.Event()

# 重置代码中的 event 对象#xff0c;使得所…线程间的通讯机制

消息队列 event 事件对象 当线程创建完成之后并不会马上执行线程而是等待某一事件发生线程才会启动 import threading# # 创建 event 对象

event threading.Event()

# 重置代码中的 event 对象使得所有该event事件都处于待命状态

event.clear()

# 阻塞线程等待 event 指令

event.wait()

# 发送 event 指令使得所有设置该 event 事件的线程执行

event.set()class MyThreading(threading.Thread):def init(self, event):super().init()self.event eventdef run(self):print(线程{}已经初始化完成随时准备启动….format(self.name))# 阻塞线程让线程等待指令后再启动self.event.wait()print({}开始执行….format(self.name))if name main:event threading.Event()# 创建 10 个自定义线程对象并放入列表threads [MyThreading(event) for i in range(10)]# 重置代码中的 event 对象使得所有该event事件都处于待命状态event.clear()# 执行线程# 执行到 run 方法中 self.event.wait() 位置即打印了线程{}已经初始化完成…[t.start() for t in threads]# 发送 event 指令使得所有设置该 event 事件的线程执行# 即启动 threads 列表中的所有线程# 接着执行 run 方法中 self.event.wait() 后面的代码即打印了{}开始执行…event.set()[t.join() for t in threads]

condition 条件对象 import threading# condition 对象适用于线程轮流执行或一个线程等待另一个线程的情况如两个人的对话等# 创建 condition 对象 cond threading.Condition()class ThreadA(threading.Thread):def init(self, cond, name):super().init(namename)self.cond conddef run(self):# 获取锁self.cond.acquire()# 线程A说了第一句话print(self.getName(), 一二三四五)# 唤醒其他处于 wait 状态的线程通知线程B可以说话了self.cond.notify()# 线程A进入 wait 状态等待线程B通知唤醒self.cond.wait()# 被线程A唤醒后说了第二句话print(self.name, 山无棱天地合乃敢与君绝)self.cond.notify() # 通知线程Bself.cond.wait() # 等待线程B通知# 被线程A唤醒后说了第三句话最后一句话print(self.name, 有钱吗借点)self.cond.notify() # 通知线程Bself.cond.release() # 释放锁class ThreadB(threading.Thread):def init(self, cond, name):super().init(namename)self.cond conddef run(self):# 获取锁self.cond.acquire()self.cond.wait() # 由于它不是第一个说话的人所以一开始等待通知# 线程B说了第一句话print(self.getName(), 上山打老虎)# 唤醒其他处于 wait 状态的线程通知线程A可以说话了self.cond.notify()# 线程B进入 wait 状态等待线程A通知唤醒self.cond.wait()# 被线程B唤醒后说了第二句话print(self.name, 海可枯石可烂激情永不散)self.cond.notify() # 通知线程Aself.cond.wait() # 等待线程A通知# 被线程B唤醒后说了第三句话最后一句话print(self.name, 没有滚)# self.cond.notify() # 已经是最后一句话不需要通知线程Aself.cond.release() # 释放锁if name main:a ThreadA(cond, AAA)b ThreadB(cond, BBB)# 线程A先说话但是不能先启动线程A# 因为如果启动了线程A然后线程A说完第一句话后通知线程B# 但是此时线程B没有启动就接收不了A的通知B就会一直处于 wait 状态即说不了话也通知不了A# A等不到B的通知也会一直处于 wait 状态# a.start()# b.start()b.start()a.start() 线程间的消息隔离机制 使用场景 在使用多线程的过程中会有一种变量的使用场景 一个变量会被所有的线程使用但是每个线程都会对该变量设置不同的值 threading.local() 提供了这种变量 使用方法 在使用多线程的过程中会有一种变量的使用场景一个变量会被所有的线程使用但是每个线程都会对该变量设置不同的值threading.local() 提供了这种变量假设有一个场景设置一个 threading.local 变量然后新建两个线程分别设置这两个 threading.local 的值再分别打印这两个 threading.local 的值看每个线程打印出来的 threading.local 值是否不一样import threading# local_data 实际上是一个对象 local_data threading.local()

设置 local_data 的名字

local_data.name local_dataclass MyThread(threading.Thread):def run(self):print(赋值前-子线程, threading.currentThread(), local_data.dict)# 在子线程中修改 local_data.name 的值local_data.name self.getName()print(赋值后-子线程, threading.currentThread(), local_data.dict)if name main:print(开始前-主线程, local_data.dict)t1 MyThread()t1.start()t1.join()t2 MyThread()t2.start()t2.join()print(结束后-主线程, local_data.dict) 输出结果 开始前-主线程 {name: local_data} 赋值前-子线程 MyThread(Thread-1, started 8480) {} 赋值后-子线程 MyThread(Thread-1, started 8480) {name: Thread-1} 赋值前-子线程 MyThread(Thread-2, started 2572) {} 赋值后-子线程 MyThread(Thread-2, started 2572) {name: Thread-2} 结束后-主线程 {name: local_data}线程池 线程池中存放多个线程当有业务需要线程来执行时可以直接从线程池中获取一个线程来执行该业务 业务执行完毕之后线程不会释放而是被放回线程池中从而节省了线程的创建以及销毁的时间。 Python concurrent.futures 模块中的 ThreadPoolExecutor 就提供了线程池该线程池有以下特点 主线程可以获取某一个线程或任务的状态以及返回值当一个线程完成的时候主线程能够立即知道让多线程和多进程的编码接口一致 线程池的简单应用 from concurrent.futures import ThreadPoolExecutor import time# 创建线程池对象并指定线程池中最大的线程数为 3

当业务数不超过 3 的时候ThreadPoolExecutor 就会创建一个新的线程来执行业务

当超过 3 时ThreadPoolExecutor 不会创建新的线程而是等待执行其他业务的线程执行完毕后返回

再将返回的线程分配给需要的业务

executor ThreadPoolExecutor(max_workers3)# 定义一个业务

假设这里模拟一个爬虫爬取一个网页页面

def get_html(timers):time.sleep(timers) # 模拟耗时操作print(获取网页信息{}完毕.format(timers))return timers# 提交要执行的函数即要完成的业务到线程池中然后线程池就会自动分配线程去完成对应的业务

submit 方法会立即返回不会阻塞主线程

get_html 的参数放在后面即 1 会作为参数传递给 get_html() 中的 timers

以下创建了四个任务

task1 executor.submit(get_html, 1) task2 executor.submit(get_html, 2) task3 executor.submit(get_html, 3) task4 executor.submit(get_html, 4)bool1 task1.done() # 检查任务是否完成完成返回 True bool2 task2.cancel() # 取消任务执行只有该任务没有被放入线程池中才能取消成功成功返回 True# 拿到任务执行的结果如 get_html 的返回值

timeout 参数用于设置等待结果的最长等待时间单位为秒

result 方法是一个阻塞方法

timers task3.result(timeout10) print(timers) print(111) 线程池中常用的方法 as_complete

线程池的简单应用

from concurrent.futures import ThreadPoolExecutor, as_completed import time# 创建线程池对象并指定线程池中最大的线程数为 3

当业务数不超过 3 的时候ThreadPoolExecutor 就会创建一个新的线程来执行业务

当超过 3 时ThreadPoolExecutor 不会创建新的线程而是等待执行其他业务的线程执行完毕后返回

再将返回的线程分配给需要的业务

executor ThreadPoolExecutor(max_workers3)# 定义一个业务

假设这里模拟一个爬虫爬取一个网页页面

def get_html(timers):time.sleep(timers) # 模拟耗时操作print(获取网页信息{}完毕.format(timers))return timers# 模拟要爬取的 url urls [1, 2, 3]

通过列表推导式构造多线程任务

all_tasks [executor.submit(get_html, url) for url in urls]

as_completed 接收一个可迭代对象

as_completed 是一个生成器当任务没有完成时它会阻塞只有当任务结束返回结果时才会继续往下执行

as_completed 函数的作用拿到所有任务执行完毕之后的结果

不需要我们手动调用 done 方法不停地判断任务是否完成

for item in as_completed(all_tasks):data item.result()print(主线程中获取任务的返回值是{}.format(data))执行结果 获取网页信息1完毕 主线程中获取任务的返回值是1 获取网页信息2完毕 主线程中获取任务的返回值是2 获取网页信息3完毕 主线程中获取任务的返回值是3map from concurrent.futures import ThreadPoolExecutor import time# 创建线程池对象并指定线程池中最大的线程数为 3

当业务数不超过 3 的时候ThreadPoolExecutor 就会创建一个新的线程来执行业务

当超过 3 时ThreadPoolExecutor 不会创建新的线程而是等待执行其他业务的线程执行完毕后返回

再将返回的线程分配给需要的业务

executor ThreadPoolExecutor(max_workers3)# 定义一个业务

假设这里模拟一个爬虫爬取一个网页页面

def get_html(timers):time.sleep(timers) # 模拟耗时操作print(获取网页信息{}完毕.format(timers))return timers# 模拟要爬取的 url urls [4, 2, 3]# map 方法和 as_complete 类似

map 也是一个生成器当任务没有完成时它会阻塞只有当任务结束返回结果时才会继续往下执行

map 会自动映射 urls 中的每一个元素传递给 get_html 函数并自动提交 不需要通过 submit 方法提交任务

map 方法直接拿到任务执行的结果

as_complete 和 map 都可以拿到线程池中各个线程执行的结果但有以下区别

as_complete 会根据任务完成的快慢得到结果即哪个任务先完成就会先得到该任务的结果

而 map 会严格按照任务的顺序得到结果比如按照 urls 列表中的映射顺序得到对应的结果

所以两种适用于不同的场景

for data in executor.map(get_html, urls):print(主线程中获取任务的返回值是{}.format(data))获取网页信息2完毕 获取网页信息3完毕 获取网页信息4完毕 主线程中获取任务的返回值是4 主线程中获取任务的返回值是2 主线程中获取任务的返回值是3wait from concurrent.futures import ThreadPoolExecutor, wait, ALL_COMPLETED, FIRST_COMPLETED import time# 创建线程池对象并指定线程池中最大的线程数为 3

当业务数不超过 3 的时候ThreadPoolExecutor 就会创建一个新的线程来执行业务

当超过 3 时ThreadPoolExecutor 不会创建新的线程而是等待执行其他业务的线程执行完毕后返回

再将返回的线程分配给需要的业务

executor ThreadPoolExecutor(max_workers3)# 定义一个业务

假设这里模拟一个爬虫爬取一个网页页面

def get_html(timers):time.sleep(timers) # 模拟耗时操作print(获取网页信息{}完毕.format(timers))return timers# 模拟要爬取的 url urls [4, 2, 3]all_tasks [executor.submit(get_html, url) for url in urls]# 让主线程阻塞直到参数里的条件成立

根据 wait 函数的参数条件成立的情况是所有任务执行完毕

ALL_COMPLETED 表示所有任务都执行完成

还有其他的参数如 FIRST_COMPLETED 表示只要有一个任务完成就条件成立

wait(all_tasks, return_whenFIRST_COMPLETED)# 如果想等代码执行完毕之后再打印下列语句可以使用 wait 语句 print(代码执行完毕) 进程池 使用 concurrent.future 模块提供的 ProcessPoolExecutor 来实现进程池用法和线程池完全一致参考上述线程池的使用建议使用该种方式使用进程池 下面是基于 Pool 类实现的进程池的使用 import multiprocessing import time# 定义一个业务

假设这里模拟一个爬虫爬取一个网页页面

def get_html(n):time.sleep(n) # 模拟耗时操作print(子进程{}获取内容成功.format(n))return nif name main:# 设置进程数一般设置为和 CPU 数量一致的比较合理# multiprocessing.cpu_count() 获取当前主机的 CPU 核心数pool multiprocessing.Pool(multiprocessing.cpu_count())# apply_async 是一个异步方法# apply 是一个同步方法# 作用类似于 submit 方法result pool.apply_async(get_html, args(2,))pool.close() # 必须在 join 方法前调用否则会抛出异常# join 方法会等待所有的子进程执行完毕之后才会继续往下执行主进程的代码# 即 join 会阻塞主进程代码pool.join()# result.get() 拿到子进程执行结果 get 方法是一个阻塞方法print(result.get())print(end…)print()# map 方法的使用pool multiprocessing.Pool(multiprocessing.cpu_count())# imap 方法会按照列表顺序输出# imap_unordered 方法则不会按照列表顺序执行而是按照任务执行的快慢输出for result in pool.imap(get_html, [1, 2, 3]):print({}休眠执行成功.format(result))线程同步信号量semaphore的使用 同步信号量的作用是用于控制同时工作的线程数量如读文件时只能同时允许两个线程读在爬虫时控制同时爬虫的线程防止触发网站反扒机制 import threading import time# 还是模拟一个爬虫

HtmlSpider 类负责根据给定的 URL 去爬取网页内容

class HtmlSpider(threading.Thread):def init(self, url, sem):super().init()self.url urlself.sem semdef run(self):time.sleep(2)print(网页内容获取完成)self.sem.release() # 线程完成任务释放锁# UrlProducer类负责给 HtmlSpider 类提供网页的 URL class UrlProducer(threading.Thread):def init(self, sem):super().init()self.sem semdef run(self):for i in range(10):self.sem.acquire() # 获取锁获取成功才能执行线程html_thread HtmlSpider(url{}.format(i), self.sem) # 创建HtmlSpider线程html_thread.start() # 启动线程if name main:# 创建线程同步信号量# 参数 value 指定允许同时工作的线程数sem threading.Semaphore(value3)url_producer UrlProducer(sem)url_producer.start()