如何为SECP256K1公钥生成的Python无限循环函数添加多线程

你的脚本属于CPU密集型任务，Python的全局解释器锁（GIL）会限制多线程在这类任务中无法利用多核，所以需要用多进程来绕过GIL，让多个核心同时工作。下面是改造后的代码，支持指定进程数（2-3个），并能准确统计总计算速度：

from time import time
from multiprocessing import Pool, Value, Lock

# SECP256K1曲线参数
a = 0
b = 7
n = 0xfffffffffffffffffffffffffffffffebaaedce6af48a03bbfd25e8cd0364141
gx = 0x79be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798
gy = 0x483ada7726a3c4655da4fbfc0e1108a8fd17b448a68554199c47d08ffb10d4b8
prime = 2**256 - 2**32 - 977

def addition(currentX, currentY, gx, gy, a, b, prime):
    if gy == 0:
        return (None, None)
    elif currentX is None and currentY is None:
        return (gx, gy)
    elif currentX == gx and currentY != gy:
        return (None, None)
    elif currentX == gx and currentY == gy and currentY == 0:
        return (None, None)
    elif currentX == gx and currentY == gy:
        s1 = (3 * pow(gx, 2, prime) + a) % prime
        s2 = (gy * 2) % prime
        s = (s1 * pow(s2, (prime - 2), prime)) % prime
        currentX = (s ** 2 - 2 * gx) % prime
        currentY = (s * (gx - currentX) - gy) % prime
    elif currentX != gx:
        s1 = (currentY - gy)
        s2 = (currentX - gx)
        s = (s1 * pow(s2, (prime - 2), prime)) % prime
        currentX = ((s ** 2) - gx - currentX) % prime
        currentY = ((s * (gx - currentX)) - gy) % prime

    return (currentX, currentY)

def secp256k1BinaryExpansion(privateKey):
    # 曲线参数作为全局变量传入，避免进程间重复传递大参数
    coef = privateKey
    currentX, currentY = gx, gy
    resultX, resultY = None, None
    while coef:
        if coef & 1:
            resultX, resultY = addition(resultX, resultY, currentX, currentY, a, b, prime)
        currentX, currentY = addition(currentX, currentY, currentX, currentY, a, b, prime)
        coef >>= 1
    return (resultX, resultY)

def worker(start_num, batch_size, counter, lock):
    """每个进程的工作函数：处理一批私钥生成任务，并更新全局计数"""
    for i in range(start_num, start_num + batch_size):
        secp256k1BinaryExpansion(i)
        with lock:
            counter.value += 1

def testParallel(process_count=2, batch_size=100):
    # 共享计数器和锁：保证多进程更新计数时不会冲突
    counter = Value('i', 0)
    lock = Lock()
    time_one = time()

    # 创建进程池，指定进程数
    with Pool(processes=process_count) as pool:
        while True:
            # 生成每个进程的起始任务编号
            current_count = counter.value
            tasks = []
            for p in range(process_count):
                start_num = current_count + p * batch_size + 1
                tasks.append(pool.apply_async(worker, args=(start_num, batch_size, counter, lock)))
            
            # 等待当前批次所有任务完成
            for task in tasks:
                task.get()
            
            # 每10秒统计一次速度
            time_two = time()
            if time_two - time_one >= 10:
                total = counter.value
                speed = total / (time_two - time_one)
                print(f"总速度: {speed:.2f} c/s")
                # 重置计时器和计数器
                time_one = time()
                with lock:
                    counter.value = 0

if __name__ == '__main__':
    # 指定进程数为2或3，对应你需要的核心占用率
    testParallel(process_count=2)

关键说明 #

• 多进程选择：multiprocessing.Pool会创建独立的进程，每个进程拥有自己的GIL，能真正利用多核CPU，适合你的CPU密集型公钥生成任务。
• 进程数设置：process_count参数设为2或3，对应你要的50%-75%核心占用（你的i5-4440是4核）。
• 共享计数：用Value和Lock实现多进程安全的计数器，避免多个进程同时更新计数导致的错误。
• 任务批量处理：每个进程处理一批任务（batch_size），减少进程间调度的开销，提升效率。
• Windows兼容性：必须加if __name__ == '__main__':保护，否则Windows下会重复启动子进程导致错误。

性能预期 #

改造后，理论上速度会接近单核心的2-3倍（比如Ubuntu下能达到600-900 c/s），核心占用率也会稳定在50%-75%左右。

内容的提问来源于stack exchange，提问作者Emma Lincoln