【Python】thread使用

1、Condition条件变量使用

# encoding:utf-8
'''Condition 提供了一种多线程通信机制，
假如线程 1 需要数据，那么线程 1 就阻塞等待，
这时线程 2 就去制造数据，线程 2 制造好数据后，
通知线程 1 可以去取数据了，然后线程 1 去获取数据。'''
import threading
import time
con = threading.Condition()
meat_num = 0def thread_consumers():  # 条件变量 condition 线程上锁con.acquire()# 全局变量声明关键字 globalglobal meat_num# 等待肉片下锅煮熟con.wait()while True:print("我来一块肉片...")meat_num -= 1print("剩余肉片数量：%d" % meat_num)time.sleep(0.5)if meat_num == 0:#  肉片吃光了，通知老板添加肉片print("老板，再来一份老肉片...")con.notify()#  肉片吃光了，等待肉片con.wait()# 条件变量 condition 线程释放锁con.release()def thread_producer():  # 条件变量 condition 线程上锁con.acquire()  # 全局变量声明关键字 globalglobal meat_num# 肉片熟了，可以开始吃了meat_num = 10print("肉片熟了，可以开始吃了...")con.notify()while True:#  阻塞函数,等待肉片吃完的通知con.wait()meat_num = 10#  添加肉片完成，可以继续开吃print("添加肉片成功！当前肉片数量：%d" % meat_num)time.sleep(1)con.notify()con.release()if __name__ == '__main__':t1 = threading.Thread(target=thread_producer)t2 = threading.Thread(target=thread_consumers)# 启动线程 -- 注意线程启动顺序，启动顺序很重要t2.start()t1.start()# 阻塞主线程，等待子线程结束t1.join()t2.join()print("程序结束！")'''
输出结果：
肉片熟了，可以开始吃了...
我来一块肉片...
剩余肉片数量：9
我来一块肉片...
剩余肉片数量：8
我来一块肉片...
剩余肉片数量：7
我来一块肉片...
剩余肉片数量：6
我来一块肉片...
剩余肉片数量：5
我来一块肉片...
剩余肉片数量：4
我来一块肉片...
剩余肉片数量：3
我来一块肉片...
剩余肉片数量：2
我来一块肉片...
剩余肉片数量：1
我来一块肉片...
剩余肉片数量：0
老板，再来一份老肉片...
添加肉片成功！当前肉片数量：10
我来一块肉片...
剩余肉片数量：9
我来一块肉片...
剩余肉片数量：8
我来一块肉片...
剩余肉片数量：7
.............
'''

2、event通信

# encoding:utf-8
"""用于线程间的通信，藉由发送线程设置的信号，
若信号为True，其他等待的线程接受到信好后会被唤醒。
提供 設置信号 event.set(), 等待信号event.wait(), 清除信号 event.clear() 。"""import threading
import time
def thread_first_job():global a# 线程进入等待状态print("Wait…")event.wait()for _ in range(3):a += 1print("This is the first thread ", a)
a = 0
# 创建event对象
event = threading.Event()
first_thread = threading.Thread(target=thread_first_job)
first_thread.start()
time.sleep(3)
# 唤醒处于等待状态的线程
print("Wake up the thread…")
event.set()
first_thread.join()
# ====== output ======
# Wait...
# Wake up the thread...
# This is the first thread  1
# This is the first thread  2
# This is the first thread  3

3、Semaphore信号量使用

# encoding:utf-8
# -*- coding:utf-8 -*-
import threading
import time
# Semaphore 跟 Lock 类似，但 Semaphore 多了计数器的功能，可以指定允许个数的线程同时执行。sem = threading.Semaphore(3)class DemoThread(threading.Thread):def run(self):print('{0} is waiting semaphore.'.format(self.name))sem.acquire()print('{0} acquired semaphore({1}).'.format(self.name, time.ctime()))time.sleep(5)print('{0} release semaphore.'.format(self.name))sem.release()if __name__ == '__main__':threads = []for i in range(4):threads.append(DemoThread(name='Thread-' + str(i)))for t in threads:t.start()for t in threads:t.join()

4、setDaemon设置守护线程

"""若希望在主线程执行完毕后，
不管其他的Thread是否已执行完毕，
都强制跟主线程一起结束，
setDaemon()必须写在start() 之前，预设为 False。"""import time
import threadingdef test():while True:print(threading.currentThread())time.sleep(1)if __name__ == '__main__':t1 = threading.Thread(target=test)t1.setDaemon(True)t1.start()t2 = threading.Thread(target=test)t2.setDaemon(True)t2.start()

5、threadPool_map使用

# encoding:utf-8
# 线程池可以控制并发多线程的数量,不会导致系统崩溃from concurrent.futures import ThreadPoolExecutor
import threading
import time# 定义一个准备作为线程任务的函数
def action(max):my_sum = 0for i in range(max):print(threading.current_thread().name + '  ' + str(i))my_sum += ireturn my_sum# 创建一个包含2条线程的线程池
# pool = ThreadPoolExecutor(max_workers=2)
# # 向线程池提交一个task, 50会作为action()函数的参数
# future1 = pool.submit(action, 50)
# # 向线程池再提交一个task, 100会作为action()函数的参数
# future2 = pool.submit(action, 100)
# # 判断future1代表的任务是否结束
# print(future1.done())
# time.sleep(3)
# # 判断future2代表的任务是否结束
# print(future2.done())
# # 查看future1代表的任务返回的结果 阻塞
# print(future1.result())
# # 查看future2代表的任务返回的结果 阻塞
# print(future2.result())
# # 关闭线程池
# pool.shutdown()# ancel()：取消该 Future 代表的线程任务。如果该任务正在执行，不可取消，则该方法返回 False；否则，程序会取消该任务，并返回 True。
# cancelled()：返回 Future 代表的线程任务是否被成功取消。
# running()：如果该 Future 代表的线程任务正在执行、不可被取消，该方法返回 True。
# done()：如果该 Funture 代表的线程任务被成功取消或执行完成，则该方法返回 True。
# result(timeout=None)：获取该 Future 代表的线程任务最后返回的结果。如果 Future 代表的线程任务还未完成，该方法将会阻塞当前线程，其中 timeout 参数指定最多阻塞多少秒。
# exception(timeout=None)：获取该 Future 代表的线程任务所引发的异常。如果该任务成功完成，没有异常，则该方法返回 None。
# add_done_callback(fn)：为该 Future 代表的线程任务注册一个“回调函数”，当该任务成功完成时，程序会自动触发该 fn 函数。
# 如果程序不希望直接调用 result() 方法阻塞线程，则可通过 Future 的 add_done_callback() 方法来添加回调函数，该回调函数形如 fn(future)。当线程任务完成后，程序会自动触发该回调函数，并将对应的 Future 对象作为参数传给该回调函数。
def spider(page):time.sleep(page)print(f"crawl task{page} finished  tread_name"+threading.currentThread().name)return page
# 该方法的功能类似于全局函数 map()，区别在于线程池的 map() 方法会为 iterables 的每个元素启动一个线程，以并发方式来执行 func 函数。这种方式相当于启动 len(iterables) 个线程，井收集每个线程的执行结果。
with ThreadPoolExecutor(max_workers=2) as pool:# map可以保证输出的顺序, submit输出的顺序是乱的.print ("start processes...")res = pool.map(spider, [0, 1, 2, 3])print("{}" .format(res))print('------done------')

6、threadPool使用

# encoding:utf-8
# 线程池可以控制并发多线程的数量,不会导致系统崩溃from concurrent.futures import ThreadPoolExecutor
import threading
import time# 定义一个准备作为线程任务的函数
def action(max):my_sum = 0for i in range(max):print(threading.current_thread().name + '  ' + str(i))my_sum += ireturn my_sum# 创建一个包含2条线程的线程池
# pool = ThreadPoolExecutor(max_workers=2)
# # 向线程池提交一个task, 50会作为action()函数的参数
# future1 = pool.submit(action, 50)
# # 向线程池再提交一个task, 100会作为action()函数的参数
# future2 = pool.submit(action, 100)
# # 判断future1代表的任务是否结束
# print(future1.done())
# time.sleep(3)
# # 判断future2代表的任务是否结束
# print(future2.done())
# # 查看future1代表的任务返回的结果 阻塞
# print(future1.result())
# # 查看future2代表的任务返回的结果 阻塞
# print(future2.result())
# # 关闭线程池
# pool.shutdown()# ancel()：取消该 Future 代表的线程任务。如果该任务正在执行，不可取消，则该方法返回 False；否则，程序会取消该任务，并返回 True。
# cancelled()：返回 Future 代表的线程任务是否被成功取消。
# running()：如果该 Future 代表的线程任务正在执行、不可被取消，该方法返回 True。
# done()：如果该 Funture 代表的线程任务被成功取消或执行完成，则该方法返回 True。
# result(timeout=None)：获取该 Future 代表的线程任务最后返回的结果。如果 Future 代表的线程任务还未完成，该方法将会阻塞当前线程，其中 timeout 参数指定最多阻塞多少秒。
# exception(timeout=None)：获取该 Future 代表的线程任务所引发的异常。如果该任务成功完成，没有异常，则该方法返回 None。
# add_done_callback(fn)：为该 Future 代表的线程任务注册一个“回调函数”，当该任务成功完成时，程序会自动触发该 fn 函数。
# 如果程序不希望直接调用 result() 方法阻塞线程，则可通过 Future 的 add_done_callback() 方法来添加回调函数，该回调函数形如 fn(future)。当线程任务完成后，程序会自动触发该回调函数，并将对应的 Future 对象作为参数传给该回调函数。with ThreadPoolExecutor(max_workers=2) as pool:# 向线程池提交一个task future_list = []for i in range(0, 2):tmp_future = pool.submit(action, (i + 1) * 50)future_list.append(tmp_future)def get_result(future):print(future.result())# 为future提供回调函数for future in future_list:future.add_done_callback(get_result)print('------done------')

7、threadingTimer

# python中使用threading.Timer作为定时函数
# 参考：https://www.pynote.net/archives/1783
# https://docs.python.org/3/library/threading.html#timer-objects
# 设置一个延迟时间，当经过了这么多时间后，某个函数被调用；如果希望反复调用，就需要编程在被调用的函数中，再次实现这个延迟一段时间调用函数的代码# threading.Timer创建的是一个线程！这个细节要注意，定时器基本上都是在线程中执行。# 如果定时器内的代码执行时间超过了定时器的间隔，怎么办？看来在python中，这个问题是不存在的，下一次定时器的计时开始，会在定时器代码执行完后才开始。
import time
import threadingdef createTimer():t = threading.Timer(2, repeat)t.start()def repeat():print('Now:', time.strftime('%H:%M:%S',time.localtime()))createTimer()createTimer()