海南省住房与城乡建设部网站,wordpress 去掉,wordpress,站酷做网站,做网站需要租服务器吗1. 线程池参数设置
CPU数量#xff1a;N线程池的核心线程数量 IO密集型的话#xff0c;一般设置为 2 * N 1#xff1b; CPU密集型的话#xff0c;一般设置为 N 1 或者 使用进程池。线程池的最大任务队列长度 #xff08;线程池的核心线程数 / 单个任务的执行时间#…1. 线程池参数设置
CPU数量N线程池的核心线程数量 IO密集型的话一般设置为 2 * N 1 CPU密集型的话一般设置为 N 1 或者 使用进程池。线程池的最大任务队列长度 线程池的核心线程数 / 单个任务的执行时间* 2 如果线程池有10个核心线程单个任务的执行时间为0.1s那么最大任务队列长度设置为200。
from concurrent.futures import ThreadPoolExecutor
thread_pool ThreadPoolExecutor(max_workers10)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数量*21cpu密集型cpu数量1
thread_pool ThreadPoolExecutor(max_workersthreadpool_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(fget 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数量*21cpu密集型cpu数量1
thread_pool ThreadPoolExecutor(max_workersthreadpool_max_worker)task_res thread_pool.map(run_task, task_params) # 批量提交任务乱序执行
print(fmain thread run finished!)
for res in task_res: # 虽然任务是乱序执行的但是得到的结果却是有序的。print(fget 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数量*21cpu密集型cpu数量1
thread_pool ThreadPoolExecutor(max_workersthreadpool_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(fmain 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_workers10)
task_param_queue queue.Queue(maxsize10)# 这里启动一个子线程一直往queue里put数据
thread_send_task Thread(targetsend_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 batchs 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_workers10)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 batchs max idx is {future_max_idx})if future_max_idx -1: # 如果为-1说明生成器的数据已经迭代完了等待kafka新生成数据print(f\n generator has empty !!!!!)time.sleep(60)
