Python3中heapq推送任务后send_instructions缺失device参数排查

问题场景 #

现有代码结构如下：

1. 定义RunThread类及全局变量 #

runner_pq = []
p_count = 0

class RunThread(threading.Thread):
    def __init__(self, func, **kwargs):
        self._func = func
        self._kwargs = kwargs
        threading.Thread.__init__(self, name="IVU")

    def run(self):
        self._func(**self._kwargs)

2. 定义任务入堆函数 #

def add_task_runner(deviceName, actionTime, func, args):
    global runner_pq
    global p_count
    t = RunThread(func, **args)
    t.name = deviceName
    task = [actionTime, p_count, t]
    p_count += 1
    heapq.heappush(runner_pq, task)

3. 添加任务的调用代码 #

RunThread.add_task_runner(device.name, device.getTime(), Tool.send_instructions, {'device': device})

4. 任务弹出执行代码 #

priority, p_count, task = heapq.heappop(runner_pq)
# 隐含后续执行 task.start()

运行时出现错误：

Exception in thread Device1:
Traceback (most recent call last):
  File "/usr/lib/python3.8/threading.py", line 932, in _bootstrap_inner
    self.run()
  File "/home/user/ivu/robot/libraries/RunThread.py", line 41, in run
    self._func(**self._kwargs)
TypeError: send_instructions() missing 1 required positional argument: 'device'

该功能在Python2.7中可正常运行，当时任务定义为task = [actionTime, t]，弹出时为priority, task = heapq.heappop(runner_pq)。

原因分析及解决方法 #

原因1：任务添加函数的调用方式错误 #

add_task_runner是独立函数，并非RunThread类的成员方法，但调用时使用了RunThread.add_task_runner(...)的方式，导致参数传递错位：

• 在Python3中，若将独立函数绑定到类属性后调用，参数会按顺序传递，但原函数定义未考虑类作为参数，导致后续参数全部偏移：
  • 原本的device.name被传给了actionTime参数
  • 原本的Tool.send_instructions被传给了args参数
  • 真正需要的{'device': device}根本未传入函数
• 最终RunThread实例的self._kwargs为空，调用send_instructions时自然缺失device参数。

解决方法：
直接调用独立函数，不要通过RunThread类调用：

add_task_runner(device.name, device.getTime(), Tool.send_instructions, {'device': device})

如果需要将add_task_runner作为RunThread的静态方法，需给函数添加@staticmethod装饰器：

class RunThread(threading.Thread):
    def __init__(self, func, **kwargs):
        self._func = func
        self._kwargs = kwargs
        threading.Thread.__init__(self, name="IVU")

    def run(self):
        self._func(**self._kwargs)

    @staticmethod
    def add_task_runner(deviceName, actionTime, func, args):
        global runner_pq
        global p_count
        t = RunThread(func, **args)
        t.name = deviceName
        task = [actionTime, p_count, t]
        p_count += 1
        heapq.heappush(runner_pq, task)

此时用RunThread.add_task_runner(...)调用即可正常传递参数。

原因2：Tool.send_instructions是实例方法却直接用类名调用 #

如果Tool.send_instructions是实例方法（未用@staticmethod或@classmethod装饰），直接通过Tool.send_instructions调用时，Python会要求传入实例作为第一个参数（self），但你的调用仅传递了device=device：

• 此时self会被错误地赋值为device，而函数需要的device参数完全缺失，导致报错。

解决方法：

1. 创建Tool实例，用实例调用方法：

   tool = Tool()
   add_task_runner(device.name, device.getTime(), tool.send_instructions, {'device': device})
   

2. 将send_instructions改为静态方法：

   class Tool:
       @staticmethod
       def send_instructions(device):
           # 方法实现
           pass
   

   之后即可直接用Tool.send_instructions调用。

关于Python2与Python3的差异 #

Python2中对未绑定方法的处理更宽松：若类的普通方法未加装饰器，用类名调用时不会强制要求传入self参数，因此参数传递不会错位；而Python3严格区分普通函数、实例方法、静态方法的调用规则，这是导致原代码在Python3中失效的核心原因之一。

内容的提问来源于stack exchange，提问作者Ben Jafari