1. 线程池参数设置
- CPU数量：
N - 线程池的核心线程数量
IO密集型的话，一般设置为2 * N + 1；
CPU密集型的话，一般设置为N + 1或者 使用进程池。 - 线程池的最大任务队列长度
（线程池的核心线程数 / 单个任务的执行时间）* 2
如果线程池有10个核心线程，单个任务的执行时间为0.1s，那么最大任务队列长度设置为200。
from concurrent.futures import ThreadPoolExecutor
thread_pool = ThreadPoolExecutor(max_workers=10)
2. submit方式提交
submit 这种提交方式是一条一条地提交任务：
1. 可以提交不同的任务函数；
2. 线程池的线程在执行任务时出现异常，程序不会停止，而且也看不到对应的报错信息；
3. 得到的结果是乱序的。
import time
from concurrent.futures import ThreadPoolExecutor, as_completeddef run_task(delay):print(f"------------> start to execute task {delay} <------------")time.sleep(delay)print(f"------------> task {delay} execute over !!! <------------")return delay + 10000task_params = [1, 4, 2, 5, 3, 6] * 10
threadpool_max_worker = 10 # io密集型：cpu数量*2+1；cpu密集型：cpu数量+1
thread_pool = ThreadPoolExecutor(max_workers=threadpool_max_worker)############################### 方式1. 虽然是异步提交任务，但是却是同步执行任务。
for p in task_params:future = thread_pool.submit(run_task, p)print(future.result()) # 直接阻塞当前线程，直到任务完成并返回结果，即变成同步############################### 方式2. 异步提交任务，而且异步执行任务，乱序执行，结果乱序。
future_list = []
for p in task_params:future = thread_pool.submit(run_task, p)future_list.append(future)for res in as_completed(future_list): # 等待子线程执行完毕，先完成的会先打印出来结果，结果是无序的print(f"get last result is {res.result()}")
3. map方式提交
map 这种提交方式可以分批次提交任务：
- 每个批次提价的任务函数都相同；
- 线程池的线程在执行任务时出现异常，程序终止并打印报错信息；
- 得到的结果是有序的。
import time
from concurrent.futures import ThreadPoolExecutor, as_completeddef run_task(delay):print(f"------------> start to execute task {delay} <------------")time.sleep(delay)print(f"------------> task {delay} execute over !!! <------------")return delay + 10000task_params = [1, 4, 2, 5, 3, 6] * 10
threadpool_max_worker = 5 # io密集型：cpu数量*2+1；cpu密集型：cpu数量+1
thread_pool = ThreadPoolExecutor(max_workers=threadpool_max_worker)task_res = thread_pool.map(run_task, task_params) # 批量提交任务，乱序执行
print(f"main thread run finished!")
for res in task_res: # 虽然任务是乱序执行的，但是得到的结果却是有序的。print(f"get last result is {res}")
4. 防止一次性提交的任务量过多
import time
from concurrent.futures import ThreadPoolExecutordef run_task(delay):print(f"------------> start to execute task <------------")time.sleep(delay)print(f"------------> task execute over !!! <------------")task_params = [1, 4, 2, 5, 3, 6] * 100
threadpool_max_worker = 10 # io密集型：cpu数量*2+1；cpu密集型：cpu数量+1
thread_pool = ThreadPoolExecutor(max_workers=threadpool_max_worker)
threadpool_max_queue_size = 200 # 线程池任务队列长度一般设置为 （线程池核心线程数/单个任务执行时间）* 2for p in task_params:print(f"*****************> 1. current queue size of thread pool is {thread_pool._work_queue.qsize()}")while thread_pool._work_queue.qsize() >= threadpool_max_queue_size:time.sleep(1) # sleep时间要超过单个任务的执行时间print(f"*****************> 2. current queue size of thread pool is {thread_pool._work_queue.qsize()}")thread_pool.submit(run_task, p)print(f"main thread run finished!")
5. 案例分享
案例背景：由于kafka一个topic的一个分区数据只能由一个消费者组中的一个消费者消费，所以现在使用线程池，从kafka里消费某一个分区的数据，将数据提取出来并存于mysql或者redis，然后手动提交offset。
import time
import queue
import concurrent.futures
from threading import Thread
from concurrent.futures import ThreadPoolExecutordef send_task_to_queue(q, params):for idx, p in enumerate(params):q.put((idx, p)) # 把kafka数据put到queue里，如果queue满了就先阻塞着，等待第15行get数据后腾出空间，这里继续put数据print(f"\n set p: {p} into task queue, queue size is {q.qsize()}")def run_task(param_queue):idx, p = param_queue.get() # 这里一直get数据，即使queue空了，只要kafka持续产生数据，第10行就会持续put数据到queue里print(f"\n ------------> start to execute task {idx} <------------")time.sleep(p)print(f"\n ------------> task {idx} execute over !!! <------------")return idxtask_params = [1, 4, 2, 5, 3] * 20 # 数据模拟kafka中消费得到的数据
thread_pool = ThreadPoolExecutor(max_workers=10)
task_param_queue = queue.Queue(maxsize=10)# 这里启动一个子线程一直往queue里put数据
thread_send_task = Thread(target=send_task_to_queue, args=(task_param_queue, task_params))
thread_send_task.start()while True: # 这里为什么一直死循环：只要生产者生产数据存储在kafka中，那么消费者就一直能获取到数据future_list = []# 分批去消费queue里的数据for i in range(10):# 这里的子线程从queue里get任务后，queue腾出空间，上面的子线程继续往里面put数据future_list.append(thread_pool.submit(run_task, task_param_queue))# 子线程任务执行结束后，从结果里取最大的索引值，可以用于redis记录，并用户手动提交offset...complete_res, uncomplete_res = concurrent.futures.wait(future_list)future_max_idx = max([future_complete.result() for future_complete in complete_res])print(f"\n ######################################## every batch's max idx is {future_max_idx}")... # 自行使用 future_max_idx 这个值做处理...6. 案例优化
上面的案例，是将数据存储于线程队列中，保证每个子线程get()到的数据不重复。
既然线程队列可以保证每个子线程get到的数据不重复，那么利用生成器的一次性特性（使用完一次就没了），是不是也能达到这个效果呢？
试着优化下：
import time
import concurrent.futures
from concurrent.futures import ThreadPoolExecutordef run_task(param_generator):try:idx, p = next(param_generator)print(f"\n ------------> start to execute task idx {idx} <------------")time.sleep(p)print(f"\n ------------> task value {p} execute over !!! <------------")return idxexcept StopIteration:return -1# 这里将task_params变成生成器，使用生成器的一次性特性：消费完一次后数据就消失了
task_params_generator = ((idx, val) for idx, val in enumerate([1, 4, 2, 5, 3] * 20))
thread_pool = ThreadPoolExecutor(max_workers=10)while True: # 这里为什么一直死循环：只要生产者生产数据存储在kafka中，那么消费者就一直能获取到数据future_list = []# 分批去消费生成器中的数据for i in range(10):# 这里的子线程从生成器中消费数据future_list.append(thread_pool.submit(run_task, task_params_generator))# 子线程任务执行结束后，从结果里取最大的索引用于redis记录，并手动提交offsetcomplete_res, uncomplete_res = concurrent.futures.wait(future_list)future_max_idx = max([future_complete.result() for future_complete in complete_res])print(f"\n ######################################## every batch's max idx is {future_max_idx}")if future_max_idx == -1: # 如果为-1，说明生成器的数据已经迭代完了，等待kafka新生成数据print(f"\n generator has empty !!!!!")time.sleep(60)
