并发编程作为现代软件开发的核心技术之一，在处理多任务执行和资源优化方面发挥着关键作用。Python作为一门功能强大的编程语言，提供了多种并发编程解决方案，其中共享内存和消息传递代表了两种截然不同的编程范式。这两种方法各有特点，适用于不同的应用场景，深入理解它们的原理和应用对于开发高性能的Python应用程序具有重要意义。

原理与实现

在Python中，multiprocessing模块提供了Value、Array和Manager等工具来实现跨进程的内存共享。这种方式的主要优势在于数据访问速度快，避免了数据复制的开销，特别适合需要频繁读写大量数据的场景。Python提供了Lock、RLock、Semaphore等同步原语来协调多个进程对共享资源的访问。

下面的代码演示了如何使用共享内存来实现一个多进程的计数器系统，展示了共享内存在协调多个进程状态方面的应用：

import random
import time
from multiprocessing import Process, Value, Array, Lock


def worker_process(process_id, shared_counter, shared_array, lock, iterations):
    """工作进程函数"""
    local_sum = 0

    for i in range(iterations):
        # 模拟一些计算工作
        work_time = random.uniform(0.001, 0.005)
        time.sleep(work_time)

        # 使用锁保护共享资源
        with lock:
            shared_counter.value += 1
            current_count = shared_counter.value

            # 更新共享数组中的进程统计
            shared_array[process_id] += 1

        local_sum += i

        if (i + 1) % 20 == 0:
            print(f"进程 {process_id}: 完成 {i + 1} 次迭代，全局计数 {current_count}")

    print(f"进程 {process_id} 完成，本地累计: {local_sum}")


def shared_memory_counter_demo():
    """
    共享内存计数器演示
    多个进程共享同一个计数器，演示共享内存的基本使用方法
    使用锁机制确保计数操作的原子性，避免竞态条件
    """
    # 创建共享内存对象
    shared_counter = Value('i', 0)  # 共享整数
    shared_array = Array('i', [0] * 4)  # 共享数组，记录每个进程的工作量
    lock = Lock()  # 同步锁

    # 创建多个工作进程
    processes = []
    iterations_per_process = 50

    print("=== 共享内存范式演示 ===")
    print(f"启动 4 个进程，每个进程执行 {iterations_per_process} 次迭代")

    start_time = time.time()

    for i in range(4):
        p = Process(
            target=worker_process,
            args=(i, shared_counter, shared_array, lock, iterations_per_process)
        )
        processes.append(p)
        p.start()

    # 等待所有进程完成
    for p in processes:
        p.join()

    end_time = time.time()

    # 输出最终结果
    print("\n=== 执行结果 ===")
    print(f"总执行时间: {end_time - start_time:.2f} 秒")
    print(f"最终计数值: {shared_counter.value}")
    print(f"预期计数值: {4 * iterations_per_process}")
    print("各进程工作量统计:")
    for i, count in enumerate(shared_array):
        print(f"  进程 {i}: {count} 次操作")

    return shared_counter.value, list(shared_array)


# 运行共享内存演示
if __name__ == '__main__':
    result = shared_memory_counter_demo()

运行结果：

=== 共享内存范式演示 ===
启动 4 个进程，每个进程执行 50 次迭代
进程 1: 完成 20 次迭代，全局计数 68
进程 3: 完成 20 次迭代，全局计数 76
进程 0: 完成 20 次迭代，全局计数 79
进程 2: 完成 20 次迭代，全局计数 93
进程 0: 完成 40 次迭代，全局计数 149
进程 1: 完成 40 次迭代，全局计数 150
进程 3: 完成 40 次迭代，全局计数 167
进程 2: 完成 40 次迭代，全局计数 173
进程 0 完成，本地累计: 1225
进程 1 完成，本地累计: 1225
进程 3 完成，本地累计: 1225
进程 2 完成，本地累计: 1225

=== 执行结果 ===
总执行时间: 1.75 秒
最终计数值: 200
预期计数值: 200
各进程工作量统计:
  进程 0: 50 次操作
  进程 1: 50 次操作
  进程 2: 50 次操作
  进程 3: 50 次操作

这个示例展示了共享内存范式的核心特征，包括数据共享、同步控制和性能优势。通过使用锁机制，确保了多个进程对共享计数器的安全访问。

消息传递范式的机制与应用

Python的multiprocessing模块提供了Queue、Pipe等通信工具来实现进程间消息传递。在消息传递模型中，每个进程维护自己独立的内存空间，通过发送和接收消息来协调工作。

以下代码实现了一个基于消息传递的任务处理系统，演示了生产者-消费者模式在实际应用中的实现方法：

import multiprocessing as mp
import time
import random
from multiprocessing import Process, Queue
import json


def producer_process(task_queue, result_queue, num_tasks):
    """生产者进程 - 生成任务"""
    print(f"生产者启动，将生成 {num_tasks} 个任务")

    for i in range(num_tasks):
        # 创建任务数据
        task = {
            'task_id': i,
            'task_type': random.choice(['compute', 'io', 'network']),
            'data': [random.randint(1, 100) for _ in range(10)],
            'priority': random.randint(1, 5),
            'created_time': time.time()
        }

        # 发送任务到队列
        task_queue.put(task)
        print(f"生产者: 创建任务 {i}, 类型: {task['task_type']}")

        # 模拟任务生成间隔
        time.sleep(random.uniform(0.1, 0.3))

    # 发送结束信号
    for _ in range(3):  # 假设有3个消费者
        task_queue.put(None)

    print("生产者: 所有任务已生成完成")


def consumer_process(consumer_id, task_queue, result_queue):
    """消费者进程 - 处理任务"""
    print(f"消费者 {consumer_id} 启动")
    processed_count = 0

    while True:
        try:
            # 从队列获取任务
            task = task_queue.get(timeout=1)

            if task is None:  # 接收到结束信号
                break

            # 处理任务
            start_time = time.time()

            # 根据任务类型执行不同的处理逻辑
            if task['task_type'] == 'compute':
                result = sum(x * x for x in task['data'])
                time.sleep(random.uniform(0.1, 0.5))  # 模拟计算时间
            elif task['task_type'] == 'io':
                result = len([x for x in task['data'] if x > 50])
                time.sleep(random.uniform(0.2, 0.4))  # 模拟IO时间
            else:  # network
                result = max(task['data']) - min(task['data'])
                time.sleep(random.uniform(0.3, 0.6))  # 模拟网络时间

            end_time = time.time()
            processing_time = end_time - start_time

            # 创建结果消息
            result_message = {
                'task_id': task['task_id'],
                'consumer_id': consumer_id,
                'result': result,
                'processing_time': processing_time,
                'task_duration': end_time - task['created_time']
            }

            # 发送结果到结果队列
            result_queue.put(result_message)
            processed_count += 1

            print(f"消费者 {consumer_id}: 完成任务 {task['task_id']}, "
                  f"结果: {result}, 处理时间: {processing_time:.3f}s")

        except Exception as e:
            print(f"消费者 {consumer_id} 发生错误: {str(e)}")
            break

    # 发送消费者完成信息
    completion_message = {
        'consumer_id': consumer_id,
        'processed_count': processed_count,
        'status': 'completed'
    }
    result_queue.put(completion_message)

    print(f"消费者 {consumer_id}: 处理完成，共处理 {processed_count} 个任务")


def result_collector_process(result_queue, expected_tasks):
    """结果收集器进程"""
    print("结果收集器启动")

    task_results = []
    consumer_stats = {}
    completed_consumers = 0

    while completed_consumers < 3:  # 等待3个消费者都完成
        try:
            result = result_queue.get(timeout=2)

            if 'status' in result and result['status'] == 'completed':
                # 收到消费者完成信息
                consumer_id = result['consumer_id']
                consumer_stats[consumer_id] = result['processed_count']
                completed_consumers += 1
                print(f"收集器: 消费者 {consumer_id} 已完成")
            else:
                # 收到任务结果
                task_results.append(result)
                print(f"收集器: 收到任务 {result['task_id']} 的结果")

        except Exception as e:
            print(f"收集器发生错误: {str(e)}")
            break

    # 分析结果
    if task_results:
        avg_processing_time = sum(r['processing_time'] for r in task_results) / len(task_results)
        avg_task_duration = sum(r['task_duration'] for r in task_results) / len(task_results)

        print("\n=== 处理结果统计 ===")
        print(f"总处理任务数: {len(task_results)}")
        print(f"平均处理时间: {avg_processing_time:.3f} 秒")
        print(f"平均任务持续时间: {avg_task_duration:.3f} 秒")
        print("消费者工作统计:")
        for consumer_id, count in consumer_stats.items():
            print(f"  消费者 {consumer_id}: {count} 个任务")

    return task_results, consumer_stats


def message_passing_demo():
    """消息传递范式演示"""
    # 创建消息队列
    task_queue = Queue(maxsize=20)  # 任务队列
    result_queue = Queue()  # 结果队列

    num_tasks = 15

    print("=== 消息传递范式演示 ===")
    print(f"启动生产者-消费者系统，处理 {num_tasks} 个任务")

    start_time = time.time()

    # 创建进程
    producer = Process(target=producer_process, args=(task_queue, result_queue, num_tasks))

    consumers = []
    for i in range(3):
        consumer = Process(target=consumer_process, args=(i, task_queue, result_queue))
        consumers.append(consumer)

    collector = Process(target=result_collector_process, args=(result_queue, num_tasks))

    # 启动所有进程
    producer.start()
    for consumer in consumers:
        consumer.start()
    collector.start()

    # 等待所有进程完成
    producer.join()
    for consumer in consumers:
        consumer.join()
    collector.join()

    end_time = time.time()
    print(f"\n总执行时间: {end_time - start_time:.2f} 秒")


if __name__ == '__main__':
    message_passing_demo()

运行结果：

=== 消息传递范式演示 ===
启动生产者-消费者系统，处理 15 个任务
消费者 0 启动
生产者启动，将生成 15 个任务
生产者: 创建任务 0, 类型: network
消费者 2 启动
消费者 1 启动
结果收集器启动
生产者: 创建任务 1, 类型: network
消费者 0: 完成任务 0, 结果: 75, 处理时间: 0.336s
收集器: 收到任务 0 的结果
生产者: 创建任务 2, 类型: compute
生产者: 创建任务 3, 类型: io
消费者 1: 完成任务 2, 结果: 37912, 处理时间: 0.311s
收集器: 收到任务 2 的结果
消费者 2: 完成任务 1, 结果: 80, 处理时间: 0.604s
生产者: 创建任务 4, 类型: compute
收集器: 收到任务 1 的结果
消费者 0: 完成任务 3, 结果: 5, 处理时间: 0.250s
收集器: 收到任务 3 的结果
生产者: 创建任务 5, 类型: compute
消费者 1: 完成任务 4, 结果: 41064, 处理时间: 0.333s
收集器: 收到任务 4 的结果
生产者: 创建任务 6, 类型: io
消费者 2: 完成任务 5, 结果: 9852, 处理时间: 0.358s
收集器: 收到任务 5 的结果
生产者: 创建任务 7, 类型: network
消费者 0: 完成任务 6, 结果: 3, 处理时间: 0.345s
收集器: 收到任务 6 的结果
生产者: 创建任务 8, 类型: io
消费者 1: 完成任务 7, 结果: 87, 处理时间: 0.571s
收集器: 收到任务 7 的结果
生产者: 创建任务 9, 类型: compute
消费者 2: 完成任务 8, 结果: 6, 处理时间: 0.367s
收集器: 收到任务 8 的结果
生产者: 创建任务 10, 类型: io
生产者: 创建任务 11, 类型: io
消费者 1: 完成任务 10, 结果: 4, 处理时间: 0.211s
收集器: 收到任务 10 的结果
消费者 0: 完成任务 9, 结果: 21822, 处理时间: 0.473s
收集器: 收到任务 9 的结果
生产者: 创建任务 12, 类型: compute
生产者: 创建任务 13, 类型: io
消费者 2: 完成任务 11, 结果: 5, 处理时间: 0.343s
收集器: 收到任务 11 的结果
生产者: 创建任务 14, 类型: network
消费者 1: 完成任务 12, 结果: 25165, 处理时间: 0.397s
收集器: 收到任务 12 的结果
消费者 0: 完成任务 13, 结果: 5, 处理时间: 0.379s
收集器: 收到任务 13 的结果
生产者: 所有任务已生成完成
消费者 1: 处理完成，共处理 5 个任务
收集器: 消费者 1 已完成
消费者 0: 处理完成，共处理 5 个任务
收集器: 消费者 0 已完成
消费者 2: 完成任务 14, 结果: 62, 处理时间: 0.345s
消费者 2: 处理完成，共处理 5 个任务
收集器: 收到任务 14 的结果
收集器: 消费者 2 已完成

=== 处理结果统计 ===
总处理任务数: 15
平均处理时间: 0.375 秒
平均任务持续时间: 0.375 秒
消费者工作统计:
  消费者 1: 5 个任务
  消费者 0: 5 个任务
  消费者 2: 5 个任务

总执行时间: 4.90 秒

这个示例展示了消息传递范式的核心概念，包括任务分发、并行处理和结果收集，体现了这种范式在复杂并发场景中的优势。

适用场景

共享内存范式在数据访问速度方面具有优势，特别适合需要频繁读写大量数据的计算密集型应用。消息传递范式虽然在通信开销方面可能略高，但其简化的编程模型和更好的错误隔离特性使其在许多场景中更为实用。

以下代码通过实际测试来比较两种范式在不同工作负载下的性能表现：

import time
import random
import time
from multiprocessing import Process, Array, Lock, Queue


def shared_memory_worker(process_id, shared_data, lock, iterations):
    """Worker function for shared memory test"""
    for i in range(iterations):
        with lock:
            shared_data[process_id] += random.randint(1, 10)
            # Simulate some computation
            temp = sum(shared_data[:])


def message_passing_worker(process_id, input_queue, output_queue):
    """Worker function for message passing test"""
    while True:
        task = input_queue.get()
        if task is None:
            output_queue.put(None)
            break

        # Process task
        result = {
            'process_id': process_id,
            'task_id': task['task_id'],
            'value': task['data']
        }
        output_queue.put(result)


def message_passing_coordinator(input_queue, output_queue, process_count, iterations):
    """Coordinator function for message passing test"""
    # Distribute tasks
    for i in range(process_count * iterations):
        task = {'task_id': i, 'data': random.randint(1, 10)}
        input_queue.put(task)

    # Send termination signals
    for _ in range(process_count):
        input_queue.put(None)

    # Collect results
    results = [0] * process_count
    completed = 0

    while completed < process_count:
        result = output_queue.get()
        if result is None:
            completed += 1
        else:
            results[result['process_id']] += result['value']

    return results


def shared_memory_test(shared_data, lock, process_count, iterations):
    """Shared memory performance test"""
    start_time = time.time()

    processes = []
    for i in range(process_count):
        p = Process(target=shared_memory_worker, args=(i, shared_data, lock, iterations))
        processes.append(p)
        p.start()

    for p in processes:
        p.join()

    end_time = time.time()
    return end_time - start_time, list(shared_data)


def message_passing_test(process_count, iterations):
    """Message passing performance test"""
    start_time = time.time()

    input_queue = Queue()
    output_queue = Queue()

    # Start coordinator
    coordinator_process = Process(
        target=message_passing_coordinator,
        args=(input_queue, output_queue, process_count, iterations)
    )
    coordinator_process.start()

    # Start workers
    workers = []
    for i in range(process_count):
        p = Process(target=message_passing_worker, args=(i, input_queue, output_queue))
        workers.append(p)
        p.start()

    coordinator_process.join()
    for p in workers:
        p.join()

    end_time = time.time()
    return end_time - start_time, None


def performance_comparison():
    """
    性能比较测试
    在相同的工作负载下比较共享内存和消息传递两种范式的性能表现
    测试包括计算密集型和通信密集型两种场景
    """
    # 测试参数
    process_count = 4
    iterations = 1000

    print("=== 并发编程范式性能比较 ===")
    print(f"测试配置: {process_count} 个进程，每个进程 {iterations} 次迭代")

    # 共享内存测试
    print("\n1. 共享内存范式测试")
    shared_data = Array('i', [0] * process_count)
    lock = Lock()

    sm_time, sm_results = shared_memory_test(shared_data, lock, process_count, iterations)
    print(f"   执行时间: {sm_time:.3f} 秒")
    print(f"   最终结果: {sm_results}")
    print(f"   总和: {sum(sm_results)}")

    # 消息传递测试
    print("\n2. 消息传递范式测试")
    mp_time, mp_results = message_passing_test(process_count, iterations)
    print(f"   执行时间: {mp_time:.3f} 秒")

    # 性能比较
    print("\n=== 性能对比结果 ===")
    print(f"共享内存方式: {sm_time:.3f} 秒")
    print(f"消息传递方式: {mp_time:.3f} 秒")

    if sm_time < mp_time:
        speedup = mp_time / sm_time
        print(f"共享内存方式快 {speedup:.2f} 倍")
    else:
        speedup = sm_time / mp_time
        print(f"消息传递方式快 {speedup:.2f} 倍")

    return sm_time, mp_time


# 运行性能比较测试
if __name__ == '__main__':
    performance_comparison()

运行结果：

=== 并发编程范式性能比较 ===
测试配置: 4 个进程，每个进程 1000 次迭代

1. 共享内存范式测试
   执行时间: 1.405 秒
   最终结果: [5506, 5655, 5428, 5420]
   总和: 22009

2. 消息传递范式测试
   执行时间: 1.291 秒

=== 性能对比结果 ===
共享内存方式: 1.405 秒
消息传递方式: 1.291 秒
消息传递方式快 1.09 倍

这个性能测试为开发者选择合适的并发编程范式提供了实际参考数据，帮助在具体应用场景中做出明智的技术决策。

最佳实践

在实际项目中选择合适的并发编程范式需要综合考虑多个因素。数据访问模式是首要考虑因素，如果应用需要频繁读写共享数据且对性能要求极高，共享内存范式可能更合适。如果应用更注重模块化、可维护性和错误隔离，消息传递范式则是更好的选择。

系统的复杂度和开发团队的技术水平也是重要考量因素。共享内存编程需要更深入的并发控制知识，消息传递编程相对更容易理解和调试。

下面的代码展示了一个实际的决策框架实现，该框架能够根据任务特征自动选择最适合的并发编程范式，并提供了混合使用策略的具体实现方法：

import multiprocessing as mp
import time
import random
from multiprocessing import Process, Value, Array, Lock, Queue
from enum import Enum
from dataclasses import dataclass
from typing import Dict, List, Any, Callable, Optional


class ConcurrencyMode(Enum):
    SHARED_MEMORY = "shared_memory"
    MESSAGE_PASSING = "message_passing"
    HYBRID = "hybrid"


@dataclass
class TaskProfile:
    """任务特征配置"""
    data_size: int  # 数据大小（字节）
    computation_intensity: float  # 计算密集度（0-1）
    communication_frequency: int  # 通信频率
    fault_tolerance_required: bool  # 是否需要容错
    scalability_required: bool  # 是否需要可扩展性


def shared_memory_worker(worker_id: int, shared_data: Array, lock: Lock,
                         task_queue: Queue, result_array: Array, tasks: List[Dict]) -> int:
    """Worker function for shared memory execution"""
    processed = 0
    while True:
        try:
            task_index = task_queue.get_nowait()
            task = tasks[task_index]

            # Execute compute-intensive task
            with lock:
                shared_data[worker_id] += task.get('value', 0)
                result_array[task_index] = task['value'] * 2

            processed += 1
            time.sleep(0.01)  # Simulate processing time

        except:
            break
    return processed


def message_passing_worker(worker_id: int, task_queue: Queue, result_queue: Queue):
    """Worker function for message passing execution"""
    while True:
        task = task_queue.get()
        if task is None:
            result_queue.put(None)
            break

        # Process task
        processed_value = task['data']['value'] * 2
        time.sleep(0.01)  # Simulate processing time

        result_queue.put({
            'index': task['index'],
            'value': processed_value,
            'worker_id': worker_id
        })


def message_passing_coordinator(task_queue: Queue, result_queue: Queue,
                                tasks: List[Dict], process_count: int):
    """Coordinator function for message passing execution"""
    # Distribute tasks
    for i, task in enumerate(tasks):
        task_queue.put({'index': i, 'data': task})

    # Send termination signals
    for _ in range(process_count):
        task_queue.put(None)

    # Collect results
    results = [0] * len(tasks)
    completed_workers = 0

    while completed_workers < process_count:
        result = result_queue.get()
        if result is None:
            completed_workers += 1
        else:
            results[result['index']] = result['value']

    return results


class ConcurrencyFramework:
    """
    并发编程范式决策框架
    根据任务特征自动选择最优的并发实现方式
    支持共享内存、消息传递和混合模式三种策略
    """

    def __init__(self):
        self.performance_history = {}
        self.decision_rules = self._initialize_decision_rules()

    def _initialize_decision_rules(self) -> Dict[str, Callable]:
        """初始化决策规则"""
        return {
            'data_size_rule': lambda profile: (
                ConcurrencyMode.SHARED_MEMORY
                if profile.data_size > 10000 and profile.communication_frequency > 50
                else ConcurrencyMode.MESSAGE_PASSING
            ),
            'computation_rule': lambda profile: (
                ConcurrencyMode.SHARED_MEMORY
                if profile.computation_intensity > 0.7
                else ConcurrencyMode.MESSAGE_PASSING
            ),
            'reliability_rule': lambda profile: (
                ConcurrencyMode.MESSAGE_PASSING
                if profile.fault_tolerance_required or profile.scalability_required
                else ConcurrencyMode.SHARED_MEMORY
            )
        }

    def analyze_task_profile(self, task_profile: TaskProfile) -> ConcurrencyMode:
        """分析任务特征并推荐并发模式"""
        scores = {mode: 0 for mode in ConcurrencyMode}

        # 应用决策规则
        for rule_name, rule_func in self.decision_rules.items():
            recommended_mode = rule_func(task_profile)
            scores[recommended_mode] += 1

        # 特殊情况处理
        if (task_profile.data_size > 5000 and
                task_profile.computation_intensity > 0.5 and
                task_profile.fault_tolerance_required):
            scores[ConcurrencyMode.HYBRID] += 2

        # 返回得分最高的模式
        return max(scores.items(), key=lambda x: x[1])[0]

    def execute_with_shared_memory(self, tasks: List[Dict], process_count: int) -> Dict:
        """使用共享内存范式执行任务"""
        # 创建共享资源
        shared_data = Array('i', [0] * process_count)
        result_array = Array('i', [0] * len(tasks))
        lock = Lock()
        task_queue = Queue()

        # 填充任务队列
        for i in range(len(tasks)):
            task_queue.put(i)

        start_time = time.time()

        # 启动工作进程
        processes = []
        for i in range(process_count):
            p = Process(
                target=shared_memory_worker,
                args=(i, shared_data, lock, task_queue, result_array, tasks)
            )
            processes.append(p)
            p.start()

        # 等待完成
        for p in processes:
            p.join()

        execution_time = time.time() - start_time

        return {
            'mode': 'shared_memory',
            'execution_time': execution_time,
            'results': list(result_array),
            'worker_totals': list(shared_data)
        }

    def execute_with_message_passing(self, tasks: List[Dict], process_count: int) -> Dict:
        """使用消息传递范式执行任务"""
        task_queue = Queue()
        result_queue = Queue()

        start_time = time.time()

        # 启动协调器
        coordinator_process = Process(
            target=message_passing_coordinator,
            args=(task_queue, result_queue, tasks, process_count)
        )
        coordinator_process.start()

        # 启动工作进程
        workers = []
        for i in range(process_count):
            p = Process(
                target=message_passing_worker,
                args=(i, task_queue, result_queue)
            )
            workers.append(p)
            p.start()

        # 等待完成并获取结果
        coordinator_process.join()
        results = [0] * len(tasks)  # Placeholder

        for p in workers:
            p.join()

        execution_time = time.time() - start_time

        return {
            'mode': 'message_passing',
            'execution_time': execution_time,
            'results': results
        }

    def execute_with_hybrid_mode(self, tasks: List[Dict], process_count: int) -> Dict:
        """使用混合模式执行任务"""
        # 将任务分为两类：计算密集型使用共享内存，IO密集型使用消息传递
        compute_tasks = [t for t in tasks if t.get('type') == 'compute']
        io_tasks = [t for t in tasks if t.get('type') == 'io']

        results = {}
        start_time = time.time()

        # 并行执行两种类型的任务
        if compute_tasks:
            sm_result = self.execute_with_shared_memory(compute_tasks, max(1, process_count // 2))
            results['compute_results'] = sm_result

        if io_tasks:
            mp_result = self.execute_with_message_passing(io_tasks, max(1, process_count // 2))
            results['io_results'] = mp_result

        execution_time = time.time() - start_time

        return {
            'mode': 'hybrid',
            'execution_time': execution_time,
            'results': results
        }

    def recommend_and_execute(self, tasks: List[Dict], task_profile: TaskProfile,
                              process_count: int = 4) -> Dict:
        """推荐并执行最优并发策略"""
        recommended_mode = self.analyze_task_profile(task_profile)

        print(f"根据任务特征分析，推荐使用: {recommended_mode.value}")

        if recommended_mode == ConcurrencyMode.SHARED_MEMORY:
            return self.execute_with_shared_memory(tasks, process_count)
        elif recommended_mode == ConcurrencyMode.MESSAGE_PASSING:
            return self.execute_with_message_passing(tasks, process_count)
        else:
            return self.execute_with_hybrid_mode(tasks, process_count)


def demonstrate_decision_framework():
    """演示决策框架的使用"""
    framework = ConcurrencyFramework()

    # 创建不同类型的测试任务
    compute_intensive_tasks = [
        {'value': random.randint(1, 100), 'type': 'compute'}
        for _ in range(50)
    ]

    io_intensive_tasks = [
        {'value': random.randint(1, 50), 'type': 'io'}
        for _ in range(30)
    ]

    mixed_tasks = compute_intensive_tasks + io_intensive_tasks

    # 定义不同的任务特征配置
    scenarios = [
        {
            'name': '计算密集型场景',
            'tasks': compute_intensive_tasks,
            'profile': TaskProfile(
                data_size=15000,
                computation_intensity=0.8,
                communication_frequency=60,
                fault_tolerance_required=False,
                scalability_required=False
            )
        },
        {
            'name': '分布式处理场景',
            'tasks': io_intensive_tasks,
            'profile': TaskProfile(
                data_size=5000,
                computation_intensity=0.3,
                communication_frequency=20,
                fault_tolerance_required=True,
                scalability_required=True
            )
        },
        {
            'name': '混合负载场景',
            'tasks': mixed_tasks,
            'profile': TaskProfile(
                data_size=8000,
                computation_intensity=0.6,
                communication_frequency=40,
                fault_tolerance_required=True,
                scalability_required=False
            )
        }
    ]

    print("=== 并发编程范式决策框架演示 ===")

    for scenario in scenarios:
        print(f"\n--- {scenario['name']} ---")
        print(f"任务数量: {len(scenario['tasks'])}")
        print(f"数据大小: {scenario['profile'].data_size} 字节")
        print(f"计算密集度: {scenario['profile'].computation_intensity}")
        print(f"通信频率: {scenario['profile'].communication_frequency}")
        print(f"容错需求: {scenario['profile'].fault_tolerance_required}")
        print(f"扩展性需求: {scenario['profile'].scalability_required}")

        result = framework.recommend_and_execute(
            scenario['tasks'],
            scenario['profile'],
            process_count=4
        )

        print(f"执行模式: {result['mode']}")
        print(f"执行时间: {result['execution_time']:.3f} 秒")

        if 'worker_totals' in result:
            print(f"工作进程统计: {result['worker_totals']}")
        if 'results' in result and isinstance(result['results'], list):
            print(f"示例结果 (前5个): {result['results'][:5]}")


if __name__ == '__main__':
    demonstrate_decision_framework()

运行结果：

=== 并发编程范式决策框架演示 ===

--- 计算密集型场景 ---
任务数量: 50
数据大小: 15000 字节
计算密集度: 0.8
通信频率: 60
容错需求: False
扩展性需求: False
根据任务特征分析，推荐使用: shared_memory
执行模式: shared_memory
执行时间: 1.789 秒
工作进程统计: [1450, 884, 0, 125]
示例结果 (前5个): [4, 126, 186, 162, 66]

--- 分布式处理场景 ---
任务数量: 30
数据大小: 5000 字节
计算密集度: 0.3
通信频率: 20
容错需求: True
扩展性需求: True
根据任务特征分析，推荐使用: message_passing
执行模式: message_passing
执行时间: 1.405 秒
示例结果 (前5个): [0, 0, 0, 0, 0]

--- 混合负载场景 ---
任务数量: 80
数据大小: 8000 字节
计算密集度: 0.6
通信频率: 40
容错需求: True
扩展性需求: False
根据任务特征分析，推荐使用: message_passing
执行模式: message_passing
执行时间: 1.490 秒
示例结果 (前5个): [0, 0, 0, 0, 0]

这个决策框架展示了如何在实际项目中系统性地选择并发编程范式，通过量化的任务特征分析来做出技术决策，确保选择的方案能够最好地满足具体应用需求。

总结

共享内存和消息传递作为并发编程的两大核心范式，各自具有独特的优势和适用场景。共享内存范式在高性能计算场景中表现卓越，而消息传递范式在分布式系统和容错应用中更具优势。Python开发中，开发者需要根据具体的应用需求、性能要求和团队技术水平来选择合适的并发策略，并可通过混合使用两种范式来充分发挥各自的技术优势。