如何提升Python中JWE解密代码的高并发处理性能？

针对高负载JWE解密场景，以下是具体的优化方案：

1. 缓存预加载的私钥与固定头对象 #

原代码每次解密都重复解析私钥、创建保护头，这是核心性能浪费点。私钥和保护头是固定资源，只需初始化一次，后续直接复用：

from jwcrypto import jwe, jwk

DATA = "eyJhbGciOiJSU0EtT0FFUC0yNTYiLCJlbmMiOiJBMjU2R0NNIiwia2lkIjoiZDgzNTIyNGItNmJlOS00MWQ0LTkyYWEtYzEyOGZkNWU5ZWNlIn0.Gadr3iydQTwKzDKp7wYeDX4gzu8_hD9t8iylgtklGdrKDOtK4pScUNuoBX92R9gnvWapNXpwcr2CEdQ45_POJ-HpFONOFCUDFuOxTfHn7ncytEDCv7a5TTbDAVBAmcH9mbkeo0DRok6vJxrprgl1USzK3UjhXzdUPTWhp8tMYtXXNzaPVbgthNqe4Hy4TX-K1Mcl_ZFcJn_JeLFE9kOWid1Dwaltzqh4pXQQoFMosumSNDGApY-DKpLpzcy_8UV890B1cnsceuggg88iPXM_GpJzKMuxkzt-451h7V500UOkdxz6M4PgAf8YLNmwziyJe3dOSEIlub0fqiZhhRNUvA.e3vioHj1RvZb58UL.-cBm-2fjYZB_EczWj78byA.3M3UqyaPUZIekudlT3aTjg"
DECRYPTION_KEY = "MIIEvgIBADANBgkqhkiG9w0BAQEFAASCBKgwggSkAgEAAoIBAQCs/yqrB8VjupUov9zApR6iEkEAPRzWrKvn+ZKlx6Rt/qTDNg0jjKahsHqTultndcB+58z+RQ/wAGGAoPm1f1JmY9P7pOGOOyGGEEsyonfa3PVwJ8Yryw3Cu0nwm2q0r9Pdv9d+BGdpPBubZ6Io5OIAtVrR7tjj8OZkGs3UXmMruXc8MTkXkO5Hccf7z+GNCu+ebNrM7sDVT+akjB2swRfHgsgwtLnnsCslyu55G6xiQIk2M7WURl7q37Y0pG4glYj4401n9lK0b8YbiuXhoeOY7799zd7gGSL1KdA14U0YSiHfbPLOV2e1Tjfv1adY97b2cSgt3ml6rbFj3siOyxyFAgMBAAECggEACu5Qlt/10/hdJmcBHtcQGXOH0ekKonHkptK5Nmpn9F9Lr94BdIaVnM/W3vC+h8oD8MDazhI1EJglkQga3yH9ZeTW5yai7ebEloD/D2L+qcCebYjF18yLwQsuf9qCIoqdm3xctGS42AhpQieJMoZL1XmgFHXAFJ8LH7Yl3U5de8ywqKdUq41YqdpklMGY3p1tS8R3R9l1x8repGyV6xQvUUsvJZ9Edkh9UBEmWG2AbEX8J5TN8h+P7Pg5dBj3sZYrcF4YvuYEIMPtfuVGqtX0SNiAWMngaYjFqG3fyfrFmxNUjF5A0GoALWF/Z4VaNtCudF4M+wptN0O77h/1gdC/PQKBgQDT1hL8LlVMRceTkZNyH8JvGN+G5m2YMsXV/GysGT3jVU2ErIvFydEhkh7fNk9jJ7vpSfIiQ2SKr+/+uh7iSa94h09AJVO5+WzcWCYa4Fen81/gaM5Hpf8bOWcQXBJRtvTN8IfjaznebYH2w2ckHLS8yZLXYSB2DGI8EaG4kgTeAwKBgQDREDAVidenvlaocYUZDNMMSVpyPCRzTIGhbEQk4JjBfdx2OKaxIXW4P4FgtH71GreJ4TFYh2BvXhpooxymM+K5bbZa+Cfhcf9fivcmXooGTsdf5hAFzIs2VXnpN82lxI9G+C4e/JDrTgADCuDGHXs3rxf5EgSYopknrtpbfAw41wKBgFdiuQQ0rNVA99plNDjg5xljXlpM1irxzvC+J6o642/TMZ4oe6RE47OAjqGvysHxky/iZDNEL8aSW2NuLe1VTkBfOQ8SFt7iCMlAx79QinhrCkMNmQKvdHQrktuJz7jLy2Lk9XmYMxARhopgAz0/ZkN2t8+Hbpej9iYnc9oqyj03AoGBAMEKM3wP1YNcK2RJvd5NiyqTKOLGSdn00NWexKVQhXxUnRbEZJa2EJ7adLo4r+nei47S0t+t0QvEVpo70p5c1oLlnTYOYbS3sFgmVuOk88mzQEF6i1zva7P15SXFkyv3fVzImL2f9KhxU217w+TrH/VgxcAENR5DNSwBNjq3JgntAoGBALYsMibwoh7PRkaqvIMpKW0bbkKIvk36MhVXcUx8Xi0UMvvd/6FBTBieWoKELAvQ4X1bL1FISHibok+/ofMAhxn9/Zt4StQYdTWtxyht6YhUFi1VKEWO1IqxLrXReIR9JAm5/gtwJBuKPh3DPw8HDlCtX8AQlh2EOInRp8hcUdPl"
KEY_ID = "d835224b-6be9-41d4-92aa-c128fd5e9ece"

def key_to_pem(key: str, is_public: bool = True) -> bytes:
    pem = [f"-----BEGIN {'PUBLIC' if is_public else 'PRIVATE'} KEY-----"]
    for i in range(0, len(key), 64):
        pem.append(key[i:i+64])
    pem.append(f"-----END {'PUBLIC' if is_public else 'PRIVATE'} KEY-----")
    return '\n'.join(pem).encode('utf-8')

# 全局预初始化，只执行一次
decrypt_key = key_to_pem(DECRYPTION_KEY, is_public=False)
PRIVATE_KEY = jwk.JWK.from_pem(decrypt_key)
PROTECTED_HEADER = {
    "alg": "RSA-OAEP-256",
    "enc": "A256GCM",
    "kid": KEY_ID,
}

def decrypt_data(encrypted_data: str) -> str:
    jwetoken = jwe.JWE(protected=PROTECTED_HEADER)
    jwetoken.deserialize(encrypted_data, key=PRIVATE_KEY)
    return jwetoken.payload.decode("utf-8")

if __name__ == '__main__':
    print(decrypt_data(DATA))

2. 直接使用底层cryptography库 #

jwcrypto是封装层，性能不如直接调用它依赖的cryptography库，减少封装开销能显著提升速度：

from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
import base64
import json

# 预加载私钥
private_key = serialization.load_pem_private_key(
    decrypt_key,
    password=None
)

def decrypt_data_fast(encrypted_data: str) -> str:
    # 拆分JWE四部分
    protected_b64, enc_key_b64, iv_b64, ciphertext_tag_b64 = encrypted_data.split('.')
    
    # URL安全Base64解码补全
    def b64url_decode(s):
        return base64.urlsafe_b64decode(s + '=' * (-len(s) % 4))
    
    # 解密对称密钥（RSA-OAEP-256）
    enc_key = b64url_decode(enc_key_b64)
    symmetric_key = private_key.decrypt(
        enc_key,
        padding.OAEP(
            mgf=padding.MGF1(algorithm=hashes.SHA256()),
            algorithm=hashes.SHA256(),
            label=None
        )
    )
    
    # AES-GCM解密
    iv = b64url_decode(iv_b64)
    ciphertext_tag = b64url_decode(ciphertext_tag_b64)
    ciphertext = ciphertext_tag[:-16]  # GCM标签固定16字节
    tag = ciphertext_tag[-16:]
    
    cipher = Cipher(algorithms.AES(symmetric_key), modes.GCM(iv, tag))
    decryptor = cipher.decryptor()
    plaintext = decryptor.update(ciphertext) + decryptor.finalize()
    
    return plaintext.decode('utf-8')

3. 并发处理高负载 #

• CPU密集场景：用多进程规避GIL限制，使用multiprocessing.Pool批量处理
• IO混合场景：用线程池或异步IO提升吞吐量

多进程示例 #

from multiprocessing import Pool

if __name__ == '__main__':
    # 模拟批量加密数据
    batch_data = [DATA] * 1000
    
    with Pool(processes=4) as pool:
        results = pool.map(decrypt_data_fast, batch_data)
    
    print(results[:5])

异步IO示例 #

import asyncio
from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=4)

async def async_decrypt(encrypted_data: str) -> str:
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(executor, decrypt_data_fast, encrypted_data)

async def main():
    batch_data = [DATA] * 1000
    tasks = [async_decrypt(data) for data in batch_data]
    results = await asyncio.gather(*tasks)
    print(results[:5])

if __name__ == '__main__':
    asyncio.run(main())

4. 启用硬件加速 #

确保安装最新版cryptography，它会自动利用OpenSSL的硬件加速特性：

pip install --upgrade cryptography

内容的提问来源于stack exchange，提问作者Ema Il