在使用 ThreadPoolExecutor线程池时,会发生 签名错误

[ymcy]

在使用单线程时,不会出现签名错误,当使用线程池去查询线程状态时,会有概率出现 签名错误的提示

2023-09-14 14:18:21,482 DEBUG   core.py line:129        Response content: {"code":"SIGN_ERROR","detail":{"detail":{"issue":"sign not match"},"field":"signature","location":"authorization","sign_information":{"method":"GET","sign_message_length":126,"truncated_sign_message":"GET\n/v3/pay/transactions/out-trade-no/*****","url":"/v3/pay/transactions/out-trade-no/******"}},"message":"签名错误，请检查后再试"}

[minibear2021]

附上代码看一下

[ymcy]

很抱歉回复晚了, 关键代码如下,该逻辑实现的是 去隔一段时间去扫描数据库中交易状态不明确的数据 python38 wechatpayv3 1.2.39

commands/scanPymentStatus.py

import sys

from django.core.management.base import BaseCommand, CommandError
from user.models import WeChatPayment
from django.db.models import QuerySet
from typing import List
from concurrent.futures import ThreadPoolExecutor
from lib.wx import api as wx_api
import datetime

class Command(BaseCommand):
    help = '扫描支付状态'
    # 最大线程数,其中的线程数量大于1,会有概念出现签名错误
    MAX_WORKERS = 4

    def add_arguments(self, parser):
        pass

    def run_retry_state(self, obj: WeChatPayment):
        """查询单个订单"""
        req_obj: WeChatPayment = wx_api.MyWeChatPay.checking_order2database(out_trade_no=obj.out_trade_no)
        # 如果查询成功,将obj换成最新的obj
        if req_obj:
            obj = req_obj
        else:
            raise ValueError("该订单最新状态查询失败")
        if obj.no_retry_state:
            """已经更改状态,无需继续扫描"""
            return
        created_at: datetime.datetime = obj.created_at.astimezone(tz=None)
        # 超过15分钟,关闭支付
        if (datetime.datetime.now().timestamp() - created_at.timestamp()) > 60 * 15:
            print(obj.out_trade_no)
            obj = wx_api.MyWeChatPay.close(out_trade_no=obj.out_trade_no, db=True)
            return obj

    def handle(self, *args, **options):
        queryset: QuerySet = WeChatPayment.objects.filter(no_retry_state=False)
        queryset: List[WeChatPayment] = list(queryset)
        with ThreadPoolExecutor(max_workers=self.MAX_WORKERS) as tp:
            tp.map(self.run_retry_state, queryset)

其中 wx_api.MyWeChatPay.checking_order2database 调用的函数

    @classmethod
    def checking_order2database(cls, transaction_id=None, out_trade_no=None, try_=5):
        """查询订单并且写入数据库"""
        # """读取数据并且写入数据库"""
        code, message = wxpay.query(transaction_id=transaction_id, out_trade_no=out_trade_no)

        if code == 401 and try_ >= 0:
            time.sleep(0.1)
            return cls.checking_order2database(transaction_id=transaction_id, out_trade_no=out_trade_no, try_=try_ - 1)

        if code != 200:
            return None
        data = json.loads(message)
        if transaction_id:
            # 如果是transaction_id查询,则更新out_trade_no
            obj = user_models.WeChatPayment.objects.get(transaction_id=transaction_id)
            obj.out_trade_no = data.get('out_trade_no')
        else:
            # 如果是则更新out_trade_no查询,则更新transaction_id
            obj = user_models.WeChatPayment.objects.get(out_trade_no=out_trade_no)
            obj.transaction_id = data.get('transaction_id')

        obj.trade_type = data.get('trade_type')
        obj.trade_state = data['trade_state']
        obj.bank_type = data.get('bank_type')
        obj.attach = str(data.get('attach'))
        # 交易完成时间
        if data.get('success_time'):
            obj.success_time = datetime.strptime(data['success_time'], "%Y-%m-%dT%H:%M:%S+08:00")

        amount = data.get('amount', {})
        if amount:
            obj.total = amount.get('total')
            obj.payer_total = amount.get('payer_total')

        payer = data.get('payer', {})
        if payer and payer.get('openid'):
            obj.openid = payer['openid']
        obj.save()
        return obj

[minibear2021]

首先微信支付订单没有『关闭』的必要，官方也没有这样的指引建议，事实上我这里遇到过大约每五个订单才实际支付一笔，其他未支付的订单完全可以不用管他。 其次如果需要更新订单状态，官方的支付消息完全值得信赖，我这里接入微信支付超过6年，每年约20万笔订单，到现在暂时没遇到过因为回调消息导致的订单处理延迟或异常。通过回调消息来处理应该是更有效率，建议将轮训查单作为备用手段。 回到这个issue上，代码作用猜测是通过多个worker同时对多笔订单查询状态，不确定是不是sdk导致你的疑问，有机会的话我会审查代码并做个验证。 在现在的状态下，确有需要的像你现在这么做的话，试下每个worker new一个wxpay，可能可以规避这个问题。

[ymcy]

好的,辛苦