運行環境

python 3.6 numpy 1.18 threadpool 1.3 內置庫：multiprocessing

代碼簡介

實現一個通用多進程+多線程支持方法。可以動態支持其他function.使調用的function具有多進程+多線程特性

代碼展示

# python3.6 import numpy as np from multiprocessing import Process, Manager import threadpool # 1.3.2class Multi_branch:"""多進程調多線程處理列表數據"""def __init__(self, func, process_num=3, thread_num=5):manager = Manager()self.func = funcself.process_num = process_numself.thread_num = thread_num# np.array_split 可以將列表平分為n組self.apportion = np.array_splitself.result_list = manager.list()# 線程執行器 將執行結果添加到進程共享列表里面def _thread_performer(self, *args, **kwargs):self.result_list.append(self.func(*args, **kwargs))# 線程構建器，將創建一個線程池def _thread_builder(self, data_list:list, *args, **kwargs):# 創建線程池pool = threadpool.ThreadPool(self.thread_num)func_var = []# 封裝 *args 和 **kwargs： func_var是一個列表，里面是很多個二元組，二元組的第一個參數是args,第二個參數是kwargsfor i in range(len(data_list)):func_var.append(((data_list[i],)+args, kwargs))# 創建線程對象，有多少個二元組，就創建多少個線程對象requests = threadpool.makeRequests(self._thread_performer, func_var)# 將線程對象放入到線程池中執行[pool.putRequest(req) for req in requests]# 等待執行完成pool.wait()def __call__(self, data_list:list, *args, **kwargs):# 按照進程數量平分數據data_list = self.apportion(data_list, self.process_num)p_list = []for i in range(self.process_num):p = Process(target=self._thread_builder, args=(list(data_list[i]),)+args, kwargs=kwargs)p.start()p_list.append(p)for p in p_list:p.join()# 返回共享列表（里面存的是運行結果）return self.result_listdef handle_text(text, *args, **kwargs):return str(text) + 'abcde'if __name__ == '__main__':texts = list(np.arange(200)) # 創建包含200個元素的列表res_list = Multi_branch(func=handle_text)(texts)print(len(res_list))for i, j in zip(texts[:5], res_list[:5]):print(i, ":", j)

結束語

還有很多可以改進的地方哦！

總結

以上是生活随笔為你收集整理的多进程多线程处理文本数据的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。