软件架构风格

简要列举了5种软件架构体系风格、包括组成、图示、示例代码。为软件系统的开发提供清晰、可预测和可维护的框架。

软件架构风格

• 说明
  • 描述特定应用领域中的系统组织方式
  • 定义
    • 一个词汇表
      • 一些构件、连接件类型
    • 一组约束
      • 指出系统如何组合构件、连接件
• 数据流体系风格
  • 批处理体系结构风格
    • Batch Processing Architecture Style
    • 原理
      • 每个处理步骤是一个单独的程序
      • 每一步须在前一步结束后开始
      • 数据须完整，整体传递
    • 组成
      • 基本构件
        • 独立的应用程序
      • 连接件
        • 某种媒介
    • 场景
      • 适用于需要处理大量、周期性的数据，且对实时性要求不高的场景
    • 图示
    • 例：
      • # 模拟初始磁带数据 data_tape = [ {"id": 1, "value": 50, "status": "valid"}, {"id": 2, "value": 30, "status": "invalid"}, {"id": 3, "value": 70, "status": "valid"}, {"id": 4, "value": 20, "status": "valid"}, ] # 1. 验证阶段 def validate_data(data_tape): print("Step 1: Validating data...") validated_data = [record for record in data_tape if record["status"] == "valid"] print(f"Validated Data: {validated_data}\n") return validated_data # 2. 排序阶段 def sort_data(data_tape): print("Step 2: Sorting data by 'value'...") sorted_data = sorted(data_tape, key=lambda x: x["value"]) print(f"Sorted Data: {sorted_data}\n") return sorted_data # 3. 更新阶段 def update_data(data_tape): print("Step 3: Updating data (incrementing 'value' by 10)...") updated_data = [{"id": record["id"], "value": record["value"] + 10, "status": record["status"]} for record in data_tape] print(f"Updated Data: {updated_data}\n") return updated_data # 4. 打印输出阶段 def print_data(data_tape): print("Step 4: Printing data...") for record in data_tape: print(f"ID: {record['id']}, Value: {record['value']}, Status: {record['status']}") print("\nProcessing complete!") # 执行批处理流程 validated_data = validate_data(data_tape) sorted_data = sort_data(validated_data) updated_data = update_data(sorted_data) print_data(updated_data)
  • 管道-过滤器体系结构风格
    • Pipe-and-Filter Architectural Style
    • 原理
      • 系统分几个序贯的处理步骤，处理数据源源不断产生的数据
      • 步骤之间通过数据流连接，一个步骤的输出是另一个步骤的输入
      • 数据传输由管道完成，处理步骤由过滤器实现
    • 组成
      • 基本构件
        • 过滤器
      • 连接件
        • 管道
    • 场景
      • 数据流场景
    • 图示
    • 例：
      • # Step 1: 转换为大写的过滤器 def to_uppercase(data): return [line.upper() for line in data] # Step 2: 过滤关键词的过滤器 def filter_keywords(data, keywords): return [line for line in data if not any(keyword in line for keyword in keywords)] # 管道执行函数 def pipeline(data, keywords): # 执行管道步骤 data = to_uppercase(data) # 转换为大写 data = filter_keywords(data, keywords) # 过滤关键词 return data # 输入数据集合 input_data = [ "This is a test.", "Debug: This line should be filtered.", "Everything is fine.", "Error occurred in the system.", ] # 需要过滤的关键词 keywords = ["ERROR", "DEBUG"] # 执行管道处理 output_data = pipeline(input_data, keywords) # 输出结果 print("Processed Data:") for line in output_data: print(line)
• 调用/返回体系结构风格
  • 原理
    • 分而治之的策略
    • 将大系统分解为若干子系统
  • 主程序/子程序风格
    • Main Program and Subroutine Style
    • 原理
      • 一般用单线程控制，把问题划分为若干处理步骤
    • 组成
      • 构件
        • 主程序、子程序
          • 子程序可合成为模块
      • 连接件
        • 调用过程
    • 例：
      • # 子程序：计算总和 def calculate_sum(numbers): return sum(numbers) # 子程序：计算最大值 def find_max(numbers): return max(numbers) # 子程序：计算最小值 def find_min(numbers): return min(numbers) # 主程序 def main(): # 输入数据 numbers = [5, 2, 9, 1, 7] # 调用子程序 total = calculate_sum(numbers) maximum = find_max(numbers) minimum = find_min(numbers) # 输出结果 print(f"Numbers: {numbers}") print(f"Sum: {total}") print(f"Max: {maximum}") print(f"Min: {minimum}") # 执行主程序 if __name__ == "__main__": main()
  • 面向对象体系结构风格
    • Object-Oriented Architectural Style
    • 原理
      • 数据的表示方法，和相应操作封装在一个抽象数据类型或对象中
    • 组成
      • 构件
        • 对象
    • 图示
    • 例：
      • # 定义一个银行账户类 class BankAccount: def __init__(self, account_number, account_holder, balance=0.0): self.account_number = account_number self.account_holder = account_holder self.balance = balance # 存款方法 def deposit(self, amount): if amount > 0: self.balance += amount print(f"Deposited {amount} to account {self.account_number}.") else: print("Invalid deposit amount.") # 取款方法 def withdraw(self, amount): if 0 < amount <= self.balance: self.balance -= amount print(f"Withdrew {amount} from account {self.account_number}.") else: print("Insufficient balance or invalid amount.") # 查询余额方法 def check_balance(self): return self.balance # 主程序 if __name__ == "__main__": # 创建账户对象 account1 = BankAccount("123456", "Alice", 1000.0) account2 = BankAccount("789012", "Bob", 500.0) # 对象交互 account1.deposit(200) account1.withdraw(500) print(f"Account {account1.account_number} balance: {account1.check_balance()}") account2.withdraw(100) print(f"Account {account2.account_number} balance: {account2.check_balance()}")
  • 层次型体系结钩风格
    • Hierarchical Architectural Style
    • 原理
      • 每层为上层提供服务，并作为下层的客户
      • 层接口只对相临层可见
    • 图示
    • 例：
      • # 数据访问层 class DataAccessLayer: def __init__(self): # 模拟数据库 self.users = {"Alice": "password123", "Bob": "qwerty"} def get_user_password(self, username): # 查询用户密码 return self.users.get(username) # 业务逻辑层 class BusinessLogicLayer: def __init__(self, dal): self.dal = dal # 依赖数据访问层 def validate_user(self, username, password): # 从数据访问层获取用户密码 stored_password = self.dal.get_user_password(username) if stored_password is None: return False, "User not found." elif stored_password == password: return True, "Login successful." else: return False, "Invalid password." # 表示层 class PresentationLayer: def __init__(self, bll): self.bll = bll # 依赖业务逻辑层 def login(self): # 获取用户输入 username = input("Enter username: ") password = input("Enter password: ") # 验证用户 success, message = self.bll.validate_user(username, password) print(message) # 主程序 if __name__ == "__main__": # 初始化各层 dal = DataAccessLayer() bll = BusinessLogicLayer(dal) pl = PresentationLayer(bll) # 启动表示层 pl.login()
  • 客户端/服务器体系结构风格
    • Client/Server Architectural Style
    • 原理
      • 基于资源不对等，为实现共享而提出
    • 分类
      • 两层 C/S 体系结构——“胖客户机、瘦服务器”
        • 组成
          • 数据库服务器
            • 数据层
          • 客户应用程序
            • 功能层、表示层
          • 网络
        • 例：
          • # 服务端 import socket def start_server(): # 创建 TCP 套接字 server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server_address = ('localhost', 8080) server_socket.bind(server_address) server_socket.listen(5) # 最大连接数 print("Server is running and waiting for connections...") while True: # 接受客户端连接 client_socket, client_address = server_socket.accept() print(f"Connected to client: {client_address}") # 接收数据 data = client_socket.recv(1024).decode('utf-8') print(f"Received from client: {data}") # 发送响应 response = f"Server received: {data}" client_socket.send(response.encode('utf-8')) # 关闭连接 client_socket.close() if __name__ == "__main__": start_server() # 客户端 import socket def start_client(): # 创建 TCP 套接字 client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server_address = ('localhost', 8080) # 连接服务器 client_socket.connect(server_address) print("Connected to server.") # 发送数据 message = input("Enter a message to send to the server: ") client_socket.send(message.encode('utf-8')) # 接收服务器响应 response = client_socket.recv(1024).decode('utf-8') print(f"Response from server: {response}") # 关闭连接 client_socket.close() if __name__ == "__main__": start_client()
      • 三层 C/S 体系结构——“瘦客户机”
        • 组成
          • 数据库服务器
            • 数据层
          • 应用服务器
            • 功能层
          • 客户应用程序
            • 表示层
          • 网络
• 以数据为中心的体系结构风格
  • 仓库体系结构风格
    • Repository Architectural Style
    • 原理
      • 系统的各个构件通过一个中央数据（即仓库）进行交互
    • 组成
      • 构件
        • 中央数据结构
          • 说明当前数据的状态
        • 独立构件
          • 操作中央数据
      • 连接件
        • 仓库与独立构件之间的交互
    • 图示
    • 例：
      • # 仓库：共享的数据存储 class Repository: def __init__(self): self.data = {} def add_student(self, student_id, student_info): if student_id in self.data: return f"Student {student_id} already exists." self.data[student_id] = student_info return f"Student {student_id} added successfully." def get_student(self, student_id): return self.data.get(student_id, "Student not found.") def update_student(self, student_id, updated_info): if student_id not in self.data: return "Student not found." self.data[student_id] = updated_info return f"Student {student_id} updated successfully." # 模块 1：数据添加模块 class AddModule: def __init__(self, repository): self.repository = repository def add_student(self, student_id, student_info): return self.repository.add_student(student_id, student_info) # 模块 2：数据查询模块 class QueryModule: def __init__(self, repository): self.repository = repository def query_student(self, student_id): return self.repository.get_student(student_id) # 模块 3：数据更新模块 class UpdateModule: def __init__(self, repository): self.repository = repository def update_student(self, student_id, updated_info): return self.repository.update_student(student_id, updated_info) # 主程序 if __name__ == "__main__": # 初始化仓库和模块 repo = Repository() add_module = AddModule(repo) query_module = QueryModule(repo) update_module = UpdateModule(repo) # 添加学生信息 print(add_module.add_student(1, {"name": "Alice", "age": 20})) print(add_module.add_student(2, {"name": "Bob", "age": 22})) # 查询学生信息 print(query_module.query_student(1)) print(query_module.query_student(3)) # 更新学生信息 print(update_module.update_student(2, {"name": "Bob", "age": 23})) print(query_module.query_student(2))
  • 黑板体系结构风格
    • Blackboard Architecture Style
    • 原理
      • 一种问题求解模型
      • 组织推理步骤，控制状态数据和问题求解之领域知识的概念框架
    • 组成
      • 黑板
        • 共享数据存储区域
        • 所有知识源都可以在黑板上读取和写入数据
        • 用于存储中间数据和最终结果
      • 知识源
        • 独立的模块或代理，每个模块包含特定的知识或技能
        • 可以是传感器、处理器、分析器等
        • 负责在特定条件下读取黑板上的数据，进行处理，并将结果写回黑板
      • 控制模块
        • 负责协调知识源的活动
        • 决定何时激活哪个知识源
        • 可以基于特定策略（如优先级、数据变化）来调度知识源
    • 场景
      • 解决负责的非结构化问题
    • 图示
    • 例：
      • # 黑板类，用于存储共享数据 class Blackboard: def __init__(self): self.data = {} def set_data(self, key, value): """在黑板上写入数据""" self.data[key] = value def get_data(self, key): """从黑板上读取数据""" return self.data.get(key) # 知识源基类 class KnowledgeSource: def __init__(self, blackboard): self.blackboard = blackboard def update(self): """更新方法，子类需实现具体逻辑""" pass # 具体的知识源实现 # 传感器 class Sensor(KnowledgeSource): def update(self): # 模拟传感器读取数据 self.blackboard.set_data('sensor_data', '传感器数据') # 处理器 class Processor(KnowledgeSource): def update(self): # 从黑板读取传感器数据并进行处理 sensor_data = self.blackboard.get_data('sensor_data') if sensor_data: processed_data = f"处理后的{sensor_data}" self.blackboard.set_data('processed_data', processed_data) # 执行器 class Actuator(KnowledgeSource): def update(self): # 从黑板读取处理后的数据并执行动作 processed_data = self.blackboard.get_data('processed_data') if processed_data: print(f"执行动作：{processed_data}") # 控制组件，负责协调知识源的更新 class Controller: def __init__(self, knowledge_sources): self.knowledge_sources = knowledge_sources def run(self): """执行知识源的更新""" for source in self.knowledge_sources: source.update() # 创建黑板 blackboard = Blackboard() # 创建知识源 sensor = Sensor(blackboard) processor = Processor(blackboard) actuator = Actuator(blackboard) # 创建控制组件并运行 controller = Controller([sensor, processor, actuator]) controller.run()
• 虚拟机体系结构风格
  • 原理
    • 人为构建一个运行环境，解析与运行自定义的语言
  • 解释器体系结构风格
    • 组成
      • 解释引擎
        • 解释
      • 存储区
        • 存储将被解释的代码
      • 数据结构
        • 记录解释引擎当前工作状态的
        • 记录源代码被解释执行进度的
    • 场景
      • 常被用来建立一种虚拟机，以弥合程序语义与硬件语义之间的差异
    • 图示
    • 例：
      • # 一个简单的计算器，解析和执行加法和减法操作 class Interpreter: def __init__(self): self.state = {} def parse(self, program): # 将程序按行分割并解析成指令列表 return [line.strip() for line in program.splitlines() if line.strip()] def execute(self, instructions): for instruction in instructions: self.execute_instruction(instruction) def execute_instruction(self, instruction): # 简单的加法和减法指令解析和执行 if instruction.startswith("ADD"): _, var, value = instruction.split() self.state[var] = self.state.get(var, 0) + int(value) elif instruction.startswith("SUB"): _, var, value = instruction.split() self.state[var] = self.state.get(var, 0) - int(value) def output(self): # 输出当前状态 for var, value in self.state.items(): print(f"{var} = {value}") # 被解释执行的程序 program = """ ADD x 5 ADD y 10 SUB x 3 """ # 创建解释器实例 interpreter = Interpreter() # 解析和执行程序 instructions = interpreter.parse(program) interpreter.execute(instructions) # 输出结果 interpreter.output()
  • 规则系统体系结构风格
    • Rule-Based System Architecture Style
    • 原理
      • 通过一组规则来推导结论或执行操作
    • 组成
      • 规则集
      • 规则解释器
      • 规则/数据选择器
      • 工作内存
    • 图示
    • 例：
      • class RuleEngine: def __init__(self): # 初始化规则引擎，包含规则列表和工作内存 self.rules = [] self.working_memory = {} def add_rule(self, condition, action): # 添加规则到规则列表，规则由条件和动作组成 self.rules.append((condition, action)) def update_memory(self, fact, value): # 更新工作内存中的事实和值 self.working_memory[fact] = value def run(self): # 运行规则引擎，检查每个规则的条件 for condition, action in self.rules: if condition(self.working_memory): # 如果条件为真，执行相应的动作 action(self.working_memory) def output(self): # 输出工作内存中的所有事实和值 for fact, value in self.working_memory.items(): print(f"{fact} = {value}") # 定义条件函数：检查温度是否大于30 def condition_temperature(memory): return memory.get('temperature', 0) > 30 # 定义动作函数：设置冷却状态为真 def action_cool_down(memory): memory['cooling'] = True # 定义条件函数：检查湿度是否大于70 def condition_humidity(memory): return memory.get('humidity', 0) > 70 # 定义动作函数：设置除湿状态为真 def action_dehumidify(memory): memory['dehumidifying'] = True # 创建规则引擎实例 engine = RuleEngine() # 添加规则：温度和湿度的条件和动作 engine.add_rule(condition_temperature, action_cool_down) engine.add_rule(condition_humidity, action_dehumidify) # 更新工作内存：设置初始温度和湿度 engine.update_memory('temperature', 35) engine.update_memory('humidity', 75) # 运行规则引擎，评估并执行适用的规则 engine.run() # 输出最终的工作内存状态 engine.output()
• 独立构件体系结构风格
  • 原理
    • 每个构件都是相对独立的个体，不直接通信
    • 降低耦合度，提升灵活性
  • 进程通信体系结构风格
    • Inter-Process Communication Architecture Style
    • 组成
      • 构件
        • 独立的（命名）过程
      • 连接件
        • 消息传递
          • 点对点
          • 异步或同步
          • 远程过程调用
    • 例：
      • from multiprocessing import Process, Queue def producer(queue): # 生产者进程：向队列中放入数据 for item in range(5): print(f"Producing {item}") queue.put(item) def consumer(queue): # 消费者进程：从队列中取出数据 while True: item = queue.get() if item is None: break print(f"Consuming {item}") if __name__ == "__main__": # 创建队列用于进程间通信 queue = Queue() # 创建生产者和消费者进程 producer_process = Process(target=producer, args=(queue,)) consumer_process = Process(target=consumer, args=(queue,)) # 启动进程 producer_process.start() consumer_process.start() # 等待生产者进程结束 producer_process.join() # 向队列发送结束信号（None）给消费者 queue.put(None) # 等待消费者进程结束 consumer_process.join()
  • 事件系统体系结构风格
    • Event System Architecture Style
    • 原理
      • 构件不直接调用一个过程，而是触发或广播一个或多个事件
      • 其他构件中的过程，在一个或多个事件中注册
      • 当一个事件被触发，系统自动调用在这个事件中注册的所有过程
    • 图示
    • 例：
      • class Event: """表示一个事件，包含名称和可选数据。""" def __init__(self, name, data=None): self.name = name self.data = data class EventManager: """管理事件的订阅和通知。""" def __init__(self): self.listeners = {} def subscribe(self, event_name, listener): """订阅一个事件的监听器。""" if event_name not in self.listeners: self.listeners[event_name] = [] self.listeners[event_name].append(listener) def notify(self, event): """通知所有监听器关于一个事件。""" if event.name in self.listeners: for listener in self.listeners[event.name]: listener(event) class EventSource: """生成和发布事件。""" def __init__(self, event_manager): self.event_manager = event_manager def publish_event(self, event_name, data=None): """创建并发布一个事件。""" event = Event(event_name, data) self.event_manager.notify(event) class EventHandler: """处理其订阅的事件。""" def __init__(self, name): self.name = name def handle_event(self, event): """处理一个事件。""" print(f"{self.name} 收到事件 '{event.name}'，数据为: {event.data}") # 示例用法 event_manager = EventManager() # 创建事件处理器 handler1 = EventHandler("处理器1") handler2 = EventHandler("处理器2") # 订阅处理器到事件 event_manager.subscribe("事件类型1", handler1.handle_event) event_manager.subscribe("事件类型2", handler2.handle_event) # 创建事件源并发布事件 event_source = EventSource(event_manager) event_source.publish_event("事件类型1", {"key": "value1"}) event_source.publish_event("事件类型2", {"key": "value2"})