项目结构：
Fan-In 模式本质：多数据源 → 统一接收器，完美匹配珠宝全流程「多环节、单数据中心」场景
架构设计
核心分层（5 层架构）
配置层（config）：全局配置、常量
消息层（message）：统一消息结构体、数据格式
业务层（process）：各业务模块（单一职责）
核心层（core）：Fan-In 队列、Sink 接收器、线程管理
启动层（main）：项目入口、编排启动
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Fan-In Pattern Fan-In（扇入）模式
# 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/19 20:47
# User : geovindu
# Product : PyCharm
# Project : pydesginpattern
# File : settings.py
"""
全局配置文件
"""
import queue
# 队列最大容量
QUEUE_MAX_SIZE = 100
# 全局 Fan-In 队列（单例）
FAN_IN_QUEUE = queue.Queue(maxsize=QUEUE_MAX_SIZE)
# 业务处理模拟时间范围（秒）
PROCESS_MIN_TIME = 0.3
PROCESS_MAX_TIME = 1.2
# 结束标识
END_SIGNAL = "SYSTEM_END"
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Fan-In Pattern Fan-In（扇入）模式
# 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/19 20:47
# User : geovindu
# Product : PyCharm
# Project : pydesginpattern
# File : business_msg.py
"""
统一业务消息结构体（企业级数据格式）
"""
from dataclasses import dataclass
from typing import Any
@dataclass
class BusinessMessage:
"""
珠宝全流程统一消息体
严格结构化：所有业务环节必须使用此格式发送数据
"""
process_name: str # 业务环节名称
task_name: str # 任务名称
task_details: Any # 任务详情
timestamp: float # 时间戳
status: str = "success" # 状态
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Fan-In Pattern Fan-In（扇入）模式
# 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/19 20:48
# User : geovindu
# Product : PyCharm
# Project : pydesginpattern
# File : fan_in_queue.py
"""
Fan-In 核心队列：线程安全、单例、全局唯一
"""
from FanInPattern.config.settings import FAN_IN_QUEUE, END_SIGNAL
from FanInPattern.message.business_msg import BusinessMessage
class FanInQueue:
"""
Fan-In 队列封装（企业级安全调用）
"""
@staticmethod
def send_msg(msg: BusinessMessage) -> None:
"""
发送业务消息
:param msg:
:return:
"""
FAN_IN_QUEUE.put(msg)
@staticmethod
def get_msg() -> BusinessMessage | str:
"""
获取消息（阻塞）
:return:
"""
return FAN_IN_QUEUE.get()
@staticmethod
def send_end_signal() -> None:
"""
发送结束信号
:return:
"""
FAN_IN_QUEUE.put(END_SIGNAL)
@staticmethod
def task_done() -> None:
"""
任务完成标识
:return:
"""
FAN_IN_QUEUE.task_done()
@staticmethod
def join() -> None:
"""
等待队列清空
:return:
"""
FAN_IN_QUEUE.join()
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Fan-In Pattern Fan-In（扇入）模式:
# 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/19 20:50
# User : geovindu
# Product : PyCharm
# Project : pydesginpattern
# File : work_sink.py
"""
工作接收器：数据中心（唯一汇聚点）
"""
import threading
import time
from FanInPattern.config.settings import END_SIGNAL
from FanInPattern.core.fan_in_queue import FanInQueue
from FanInPattern.message.business_msg import BusinessMessage
class WorkSink:
"""
珠宝业务数据中心（Fan-In Sink）
"""
def __init__(self):
self.total_count = 0
self.process_stats = {} # 各环节统计
def _update_stats(self, process_name: str) -> None:
"""
更新业务统计
:param process_name:
:return:
"""
self.process_stats[process_name] = self.process_stats.get(process_name, 0) + 1
def run(self) -> None:
"""
启动接收器
:return:
"""
print("=" * 90)
print("🏢 【企业级】珠宝业务数据中心已启动 - Fan-In 工作接收器")
print("=" * 90)
while True:
data = FanInQueue.get_msg()
# 结束信号
if data == END_SIGNAL:
self._print_final_report()
FanInQueue.task_done()
break
# 处理业务消息
if isinstance(data, BusinessMessage):
self._handle_message(data)
FanInQueue.task_done()
time.sleep(0.1)
def _handle_message(self, msg: BusinessMessage) -> None:
"""
处理单条消息
:param msg:
:return:
"""
self.total_count += 1
self._update_stats(msg.process_name)
print(
f"📊 接收 | {msg.process_name:10s} | "
f"任务：{msg.task_name:15s} | 状态：{msg.status}"
)
def _print_final_report(self) -> None:
"""
打印企业级汇总报告
:return:
"""
print("\n" + "=" * 90)
print("📈 珠宝全流程业务汇总报告")
print("=" * 90)
print(f"✅ 总处理任务数：{self.total_count}")
for process, count in self.process_stats.items():
print(f" 「{process}」：{count} 项")
print("=" * 90)
# 单例接收器（全局唯一）
SINK_INSTANCE = WorkSink()
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Fan-In Pattern Fan-In（扇入）模式:
# 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/19 20:52
# User : geovindu
# Product : PyCharm
# Project : pydesginpattern
# File : base_process.py
"""
业务基类：所有业务环节必须继承此类
"""
import abc
import time
import random
import threading
from FanInPattern.config.settings import PROCESS_MIN_TIME, PROCESS_MAX_TIME
from FanInPattern.core.fan_in_queue import FanInQueue
from FanInPattern.message.business_msg import BusinessMessage
class BaseProcess(abc.ABC):
"""
抽象业务基类
职责：统一执行逻辑、发送消息、模拟耗时
"""
def __init__(self, process_name: str):
self.process_name = process_name
self.task_list = []
@abc.abstractmethod
def init_tasks(self) -> None:
"""
初始化任务（子类必须实现）
:return:
"""
pass
def _simulate_process(self) -> None:
"""
模拟业务处理耗时
:return:
"""
sleep_time = random.uniform(PROCESS_MIN_TIME, PROCESS_MAX_TIME)
time.sleep(sleep_time)
def _send_task_msg(self, task: str) -> None:
"""
构造并发送消息
:param task:
:return:
"""
msg = BusinessMessage(
process_name=self.process_name,
task_name=task,
task_details=task,
timestamp=time.time()
)
FanInQueue.send_msg(msg)
def execute(self) -> None:
"""
执行业务流程
:return:
"""
self.init_tasks()
for task in self.task_list:
self._simulate_process()
self._send_task_msg(task)
def start_thread(self) -> threading.Thread:
"""
启动独立线程执行
:return:
"""
thread = threading.Thread(target=self.execute, name=self.process_name)
thread.start()
return thread
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Fan-In Pattern Fan-In（扇入）模式:
# 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/19 21:09
# User : geovindu
# Product : PyCharm
# Project : pydesginpattern
# File : design.py
from FanInPattern.process.base_process import BaseProcess
class DesignProcess(BaseProcess):
"""
珠宝设计
"""
def __init__(self):
super().__init__("珠宝设计")
def init_tasks(self):
"""
:return:
"""
self.task_list = ["钻戒款式设计", "项链3D建模", "手镯图纸审核"]
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Fan-In Pattern Fan-In（扇入）模式:
# 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/19 21:11
# User : geovindu
# Product : PyCharm
# Project : pydesginpattern
# File : purchase.py
from FanInPattern.process.base_process import BaseProcess
class PurchaseProcess(BaseProcess):
"""
原料采购
"""
def __init__(self):
super().__init__("原料采购")
def init_tasks(self):
"""
:return:
"""
self.task_list = [
"采购1克拉南非钻石", "采购999足金500g", "采购红宝石10颗"
]
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：
# 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/19 21:23
# User : geovindu
# Product : PyCharm
# Project : pydesginpattern
# File : produce.py
from FanInPattern.process.base_process import BaseProcess
class ProduceProcess(BaseProcess):
"""
生产加工
"""
def __init__(self):
super().__init__("生产加工")
def init_tasks(self):
"""
:return:
"""
self.task_list = ["钻石镶嵌加工", "黄金手镯抛光", "金饰铸造成型"]
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Fan-In Pattern Fan-In（扇入）模式
# 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/19 21:26
# User : geovindu
# Product : PyCharm
# Project : pydesginpattern
# File : quality.py
from FanInPattern.process.base_process import BaseProcess
class QualityProcess(BaseProcess):
"""
质量检测
"""
def __init__(self):
super().__init__("质量检测")
def init_tasks(self):
self.task_list = ["黄金纯度检测", "钻石工艺检测", "珠宝鉴定证书出具"]
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Fan-In Pattern Fan-In（扇入）模式
# 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/19 21:28
# User : geovindu
# Product : PyCharm
# Project : pydesginpattern
# File : warehouse.py
from FanInPattern.process.base_process import BaseProcess
class WarehouseProcess(BaseProcess):
"""
仓储管理
"""
def __init__(self):
super().__init__("仓储管理")
def init_tasks(self):
self.task_list = ["钻戒入库登记", "黄金库存盘点", "宝石库存预警"]
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Fan-In Pattern Fan-In（扇入）模式
# 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/19 21:28
# User : geovindu
# Product : PyCharm
# Project : pydesginpattern
# File : sale.py
from FanInPattern.process.base_process import BaseProcess
class SaleProcess(BaseProcess):
"""
销售环节
"""
def __init__(self):
super().__init__("销售环节")
def init_tasks(self):
self.task_list = ["线上钻戒售出", "门店黄金手镯售出", "珠宝批发订单发货"]
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Fan-In Pattern Fan-In（扇入）模式
# 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/19 21:29
# User : geovindu
# Product : PyCharm
# Project : pydesginpattern
# File : after_sale.py
from FanInPattern.process.base_process import BaseProcess
class AfterSaleProcess(BaseProcess):
"""
售后服务
"""
def __init__(self):
super().__init__("售后服务")
def init_tasks(self):
self.task_list = ["钻戒免费清洗", "项链维修", "黄金首饰保养"]
业务调用：
# encoding: utf-8
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述： Fan-In Pattern Fan-In（扇入）模式
# 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/19 21:51
# User : geovindu
# Product : PyCharm
# Project : pydesginpattern
# File : FanInBll.py
"""
企业级编排
"""
import threading
from FanInPattern.core.fan_in_queue import FanInQueue
from FanInPattern.core.work_sink import SINK_INSTANCE
# 导入所有业务流程
from FanInPattern.process.purchase import PurchaseProcess
from FanInPattern.process.design import DesignProcess
from FanInPattern.process.produce import ProduceProcess
from FanInPattern.process.quality import QualityProcess
from FanInPattern.process.warehouse import WarehouseProcess
from FanInPattern.process.sale import SaleProcess
from FanInPattern.process.after_sale import AfterSaleProcess
class JewelryBusinessOrchestration(object):
"""
业务流程编排器
"""
@staticmethod
def get_all_processes() -> list:
"""
获取所有业务环节（扩展只需在此添加）
:return:
"""
return [
PurchaseProcess(),
DesignProcess(),
ProduceProcess(),
QualityProcess(),
WarehouseProcess(),
SaleProcess(),
AfterSaleProcess()
]
@staticmethod
def run():
"""启动全流程"""
# 1. 启动接收器线程
sink_thread = threading.Thread(target=SINK_INSTANCE.run, name="WorkSink")
sink_thread.start()
# 2. 启动所有业务线程
processes = JewelryBusinessOrchestration.get_all_processes()
threads = [p.start_thread() for p in processes]
# 3. 等待所有业务完成
for t in threads:
t.join()
# 4. 发送结束信号
FanInQueue.send_end_signal()
# 5. 等待队列处理完成
FanInQueue.join()
sink_thread.join()
class FanInBll(object):
"""
"""
def demo(self):
"""
:return:
"""
print("🚀 企业级珠宝 Fan-In 业务系统启动...\n")
JewelryBusinessOrchestration.run()
print("\n🎉 系统全部执行完成！")
输出：
哲学管理(学)人生, 文学艺术生活, 自动(计算机学)物理(学)工作, 生物(学)化学逆境, 历史(学)测绘(学)时间, 经济(学)数学金钱(理财), 心理(学)医学情绪, 诗词美容情感, 美学建筑(学)家园, 解构建构(分析)整合学习, 智商情商(IQ、EQ)运筹(学)生存.---Geovin Du(涂聚文）
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。