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

python: Parallelism Pattern

news 2026/6/18 11:12:26

项目结构：

# encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎 # 描述：Parallelism Pattern 并行模式 # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/6/17 20:44 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : settings.py import os # 基础路径 BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) IMAGE_SAVE_DIR = os.path.join(BASE_DIR, "jewelry_images") # 并行配置 THREAD_MAX_WORKERS = 16 PROCESS_MAX_WORKERS = 4 # 业务模拟配置 JEWELRY_COUNT = 100 ORDER_COUNT = 50 WAREHOUSE_COUNT = 5 # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎 # 描述：Parallelism Pattern 并行模式 # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/6/17 20:46 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : entities.py from dataclasses import dataclass from ParallelismPattern.config.settings import JEWELRY_COUNT, ORDER_COUNT, WAREHOUSE_COUNT @dataclass class Jewelry(object): id: str @dataclass class Order(object): id: str @dataclass class Warehouse(object): id: str class EntityFactory(object): """ 实体工厂：统一生成测试数据 """ @staticmethod def get_jewelries(): """ :return: """ return [Jewelry(f"珠宝_{i:03d}") for i in range(1, JEWELRY_COUNT + 1)] @staticmethod def get_orders(): """ :return: """ return [Order(f"订单_{i:03d}") for i in range(1, ORDER_COUNT + 1)] @staticmethod def get_warehouses(): """ :return: """ return [Warehouse(f"仓库_{chr(65 + i)}") for i in range(WAREHOUSE_COUNT)] # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎 # 描述：Parallelism Pattern 并行模式 # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/6/17 20:44 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : logger.py import logging import os from datetime import datetime from ParallelismPattern.config.settings import BASE_DIR # 日志目录 LOG_DIR = os.path.join(BASE_DIR, "logs") os.makedirs(LOG_DIR, exist_ok=True) # 日志格式 logging.basicConfig( level=logging.INFO, format="%(asctime)s | %(levelname)s | %(module)s | %(message)s", handlers=[ logging.FileHandler(os.path.join(LOG_DIR, f"jewelry_{datetime.now().date()}.log"), encoding="utf-8"), logging.StreamHandler() ] ) logger = logging.getLogger("jewelry_system") # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎 # 描述：Parallelism Pattern 并行模式 # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/6/17 20:44 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : parallel.py from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor, as_completed from ParallelismPattern.utils.logger import logger from ParallelismPattern.config.settings import THREAD_MAX_WORKERS, PROCESS_MAX_WORKERS class ParallelExecutor(object): """ 通用并行执行器，支持线程/进程，完全解耦业务 并行执行器（核心通用组件） """ @staticmethod def run_thread_tasks(task_func, task_list, max_workers=THREAD_MAX_WORKERS): """ 执行多线程任务（IO密集） :param task_func: :param task_list: :param max_workers: :return: """ return ParallelExecutor._run_tasks( executor_cls=ThreadPoolExecutor, task_func=task_func, task_list=task_list, max_workers=max_workers ) @staticmethod def run_process_tasks(task_func, task_list, max_workers=PROCESS_MAX_WORKERS): """ 执行多进程任务（CPU密集） :param task_func: :param task_list: :param max_workers: :return: """ return ParallelExecutor._run_tasks( executor_cls=ProcessPoolExecutor, task_func=task_func, task_list=task_list, max_workers=max_workers ) @staticmethod def _run_tasks(executor_cls, task_func, task_list, max_workers): """ 内部通用执行逻辑 :param executor_cls: :param task_func: :param task_list: :param max_workers: :return: """ results = [] try: with executor_cls(max_workers=max_workers) as executor: future_map = {executor.submit(task_func, item): item for item in task_list} for future in as_completed(future_map): try: res = future.result() results.append(res) logger.info(res) except Exception as e: task_item = future_map[future] err_msg = f"任务执行失败 | 函数={task_func.__name__} | 项={task_item} | 错误={str(e)}" logger.error(err_msg) results.append(err_msg) except Exception as e: logger.error(f"并行执行器异常：{str(e)}") return results

# encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎 # 描述：Parallelism Pattern 并行模式 # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/6/17 20:46 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : quality_service.py import time import random class QualityService(object): """ 质检服务 """ @staticmethod def inspect(jewelry): """ :param jewelry: :return: """ time.sleep(0.2) result = random.choice(["合格", "不合格"]) return f"【质检】{jewelry.id}：{result}" # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎 # 描述：Parallelism Pattern 并行模式 # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/6/17 20:50 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : image_service.py import os from PIL import Image, ImageDraw from ParallelismPattern.config.settings import IMAGE_SAVE_DIR from ParallelismPattern.utils.logger import logger import time class ImageService(object): """ 图片服务 """ @staticmethod def process(jewelry): """ :param jewelry: :return: """ try: os.makedirs(IMAGE_SAVE_DIR, exist_ok=True) img = Image.new('RGB', (200, 200), color='gold') draw = ImageDraw.Draw(img) draw.text((10, 10), jewelry.id, fill='black') img.save(os.path.join(IMAGE_SAVE_DIR, f"{jewelry.id}.jpg")) time.sleep(0.15) return f"【图片处理】{jewelry.id}：完成" except Exception as e: logger.error(f"图片处理失败 {jewelry.id}: {e}") return f"【图片处理】{jewelry.id}：失败" # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎 # 描述：Parallelism Pattern 并行模式 # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/6/17 20:54 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : stock_service.py import time import random class StockService(object): """ 库存服务 """ @staticmethod def check(warehouse): """ :param warehouse: :return: """ time.sleep(0.3) num = random.randint(50, 200) return f"【库存盘点】{warehouse.id}：库存{num}件" # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎 # 描述：Parallelism Pattern 并行模式 # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/6/17 20:55 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : notice_service.py import time class NoticeService(object): """ 通知服务 """ @staticmethod def send_ship(order): time.sleep(0.1) return f"【发货通知】{order.id}：已推送短信+邮件" # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎 # 描述：Parallelism Pattern 并行模式 # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/6/17 20:57 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : cert_service.py import time import random class CertService(object): """ 证书服务 """ @staticmethod def verify(jewelry): time.sleep(0.25) cert = f"NGTC{random.randint(100000,999999)}" return f"【证书核验】{jewelry.id}：证书{cert} 有效" # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎 # 描述： Parallelism Pattern 并行模式 # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/6/17 21:06 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : price_service.py import time import random class PriceService(object): """ 定价服务 """ @staticmethod def calculate(jewelry): time.sleep(0.22) price = random.randint(1000, 50000) return f"【定价核算】{jewelry.id}：定价{price}元" # encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎 # 描述：Parallelism Pattern 并行模式 # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/6/17 21:07 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : task_manager.py from ParallelismPattern.models.entities import EntityFactory from ParallelismPattern.services.quality_service import QualityService from ParallelismPattern.services.image_service import ImageService from ParallelismPattern.services.stock_service import StockService from ParallelismPattern.services.notice_service import NoticeService from ParallelismPattern.services.cert_service import CertService from ParallelismPattern.services.price_service import PriceService from ParallelismPattern.utils.parallel import ParallelExecutor class JewelryTaskManager(object): """ 任务编排中心 新增业务：只需要在这里加一个方法 + 一个服务，无需动并行核心 """ def __init__(self): """ """ self.jewelries = EntityFactory.get_jewelries() self.orders = EntityFactory.get_orders() self.warehouses = EntityFactory.get_warehouses() def run_all_tasks(self): """ 执行全流程任务 :return: """ # IO密集 → 线程池 ParallelExecutor.run_thread_tasks(NoticeService.send_ship, self.orders) ParallelExecutor.run_thread_tasks(StockService.check, self.warehouses) ParallelExecutor.run_thread_tasks(ImageService.process, self.jewelries) # CPU密集 → 进程池 ParallelExecutor.run_process_tasks(QualityService.inspect, self.jewelries) ParallelExecutor.run_process_tasks(CertService.verify, self.jewelries) ParallelExecutor.run_process_tasks(PriceService.calculate, self.jewelries)

调用：

# encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎 # 描述：Parallelism Pattern 并行模式 # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/6/17 21:09 # User : geovindu # Product : PyCharm # Project : pydesginpattern # File : ParallelismBll.py import time from ParallelismPattern.tasks.task_manager import JewelryTaskManager from ParallelismPattern.utils.logger import logger class ParallelismBll(object): """ """ def demo(self): """ :return: """ logger.info("=== 企业级珠宝并行任务系统 启动 ===") start = time.time() # 启动任务 manager = JewelryTaskManager() manager.run_all_tasks() end = time.time() logger.info(f"=== 系统执行完成 | 总耗时：{end - start:.2f}s ===")

输出：

查看全文

http://www.jsqmd.com/news/1035431/