多线程统计0-9随机向量计数问题及代码报错求助

问题核心 #

你需要完成以下任务：

• 生成10^8个0-9的随机数字向量
• 统计每个数字的出现次数，实时显示统计进度
• 支持1/2/5/10个线程执行统计，用信号量或锁实现互斥
• 每种线程数+互斥策略组合重复30次实验，分析平均时长与置信区间

你之前遇到的thread.get_result()报错，是因为Python标准库threading.Thread并没有这个方法——自定义线程类如果没正确实现结果返回逻辑，就会出现这类问题。下面是完整的可运行实现方案。

实现思路 #

1. 高效数据生成：用numpy生成int8类型的随机数组，大幅降低内存占用（10^8个int8仅占约100MB）。
2. 线程协作策略：采用分块局部统计+结果合并的方式，每个线程独立统计自己负责的数据块，最后主线程合并结果——这种方式能最小化互斥锁的使用频率，提升性能。
3. 实时进度显示：用共享计数器记录已处理数据量，通过锁/信号量保护更新操作，主线程定时打印进度。
4. 互斥实现：分别用threading.Lock和threading.Semaphore(1)实现互斥（两者在单许可场景下功能等价）。
5. 性能测试：重复30次实验，计算平均时长、标准差和95%置信区间（基于t分布）。

完整代码实现 #

import threading
import numpy as np
import time
import statistics
import math

# 生成10^8个0-9的随机数字向量
def generate_data(size=10**8):
    return np.random.randint(0, 10, size=size, dtype=np.int8)

# 线程处理函数：统计指定数据块的局部频次，更新进度
def count_chunk(data, start, end, local_counts, progress_mutex, processed):
    local_count = [0] * 10
    for idx in range(start, end):
        num = data[idx]
        local_count[num] += 1
        # 每处理10000条数据更新一次进度，减少锁竞争
        if (idx - start + 1) % 10000 == 0:
            with progress_mutex:
                processed[0] += 10000
    # 将局部结果存入共享列表
    local_counts.append(local_count)

# 单次统计任务执行函数
def run_statistics(data, num_threads, mutex_type='lock'):
    total_size = len(data)
    chunk_size = total_size // num_threads
    local_counts = []
    processed = [0]  # 用列表实现可变共享变量
    threads = []

    # 选择互斥机制
    if mutex_type == 'lock':
        progress_mutex = threading.Lock()
    elif mutex_type == 'semaphore':
        progress_mutex = threading.Semaphore(1)
    else:
        raise ValueError("互斥类型仅支持'lock'或'semaphore'")

    # 创建并启动线程
    for i in range(num_threads):
        start = i * chunk_size
        # 最后一个线程处理剩余所有数据
        end = start + chunk_size if i != num_threads - 1 else total_size
        thread = threading.Thread(
            target=count_chunk,
            args=(data, start, end, local_counts, progress_mutex, processed)
        )
        threads.append(thread)

    start_time = time.time()
    for thread in threads:
        thread.start()

    # 实时打印统计进度
    while processed[0] < total_size:
        print(f"\r已统计: {processed[0]}/{total_size} ({processed[0]/total_size*100:.2f}%)", end='')
        time.sleep(0.1)
    print(f"\r已统计: {total_size}/{total_size} (100.00%)")

    # 等待所有线程结束
    for thread in threads:
        thread.join()

    # 合并所有局部统计结果
    final_counts = [0] * 10
    for count in local_counts:
        for num in range(10):
            final_counts[num] += count[num]

    return final_counts, time.time() - start_time

# 性能测试：重复30次实验，计算统计指标
def performance_test(data, num_threads, mutex_type):
    duration_list = []
    for idx in range(30):
        _, duration = run_statistics(data, num_threads, mutex_type)
        duration_list.append(duration)
        print(f"第{idx+1}次实验完成，耗时: {duration:.2f}s")

    # 计算平均时长、标准差和95%置信区间
    mean_duration = statistics.mean(duration_list)
    std_dev = statistics.stdev(duration_list)
    # 自由度29的t分布95%分位数约为2.045
    margin_error = 2.045 * (std_dev / math.sqrt(30))
    conf_interval = (mean_duration - margin_error, mean_duration + margin_error)

    return {
        "平均时长(s)": round(mean_duration, 2),
        "标准差(s)": round(std_dev, 2),
        "95%置信区间(s)": (round(conf_interval[0], 2), round(conf_interval[1], 2))
    }

if __name__ == "__main__":
    # 预生成数据（仅生成一次，避免实验误差）
    print("正在生成10^8个随机数字...")
    data = generate_data()
    print("数据生成完成！\n")

    # 测试所有线程数+互斥策略组合
    thread_counts = [1, 2, 5, 10]
    mutex_types = ['lock', 'semaphore']
    test_results = {}

    for mutex in mutex_types:
        test_results[mutex] = {}
        for threads in thread_counts:
            print(f"=== 测试配置：{threads}线程 + {mutex}互斥 ===")
            perf_result = performance_test(data, threads, mutex)
            test_results[mutex][threads] = perf_result
            print(f"平均时长: {perf_result['平均时长(s)']}s")
            print(f"95%置信区间: {perf_result['95%置信区间(s)']}\n")

    # 打印最终数字统计结果（以10线程+锁为例）
    print("=== 最终数字频次统计结果 ===")
    final_counts, _ = run_statistics(data, 10, 'lock')
    for num in range(10):
        print(f"数字{num}: {final_counts[num]}次")

关键细节说明 #

1. 结果获取方式：通过共享列表local_counts收集每个线程的局部统计结果，线程结束后主线程合并——替代了不存在的get_result()方法，逻辑更可靠。
2. 互斥机制差异：Lock是专用互斥锁，Semaphore(1)是单许可信号量，两者在互斥场景下性能几乎无差异；如果需要多线程同时访问资源，可调整信号量许可数。
3. 性能优化点：
   • 用局部计数减少锁竞争，仅在更新进度时使用互斥。
   • 批量更新进度（每10000条一次），避免频繁锁操作拖慢速度。
   • 预生成数据，避免每次实验重复生成带来的时间误差。

预期性能分析 #

• 线程数从1增加到CPU核心数时，耗时会显著降低；超过核心数后，线程切换开销增加，耗时会趋于稳定甚至略有上升。
• Lock和Semaphore的性能差异极小，测试中几乎无法区分。
• 30次重复实验的置信区间能有效反映结果的稳定性，区间越窄说明性能波动越小。

内容的提问来源于stack exchange，提问作者Davyson Ribeiro