python - 如何在 Python 中对特定函数的一个实例进行多线程/多处理？

Question

我正在运行一个控制机器人的 Python 脚本，但我对如何对电机控制函数进行多线程处理感到有点困惑。

问题是硬件的设计使得电机不会移动，除非在电机控制功能中有几次休眠，因为硬件需要时间来向电机发送电信号。由于电机控制功能中的这些休眠，整个程序暂停并停止读取传感器数据。

我想做的是知道如何在调用电机控制函数后对其进行多线程/多处理，但是一旦程序在循环的后续迭代中再次调用，它会检查电机控制是否是仍在运行(即它没有完成 sleep )。如果它仍在运行，它会跳过电机控制调用并继续循环，读取传感器数据，然后再次检查电机控制功能是否仍在运行。当然，如果电机控制功能不再运行，我希望再次调用它。

基本上，整个程序只需要两个线程:一个运行主程序，另一个在每次电机控制功能完成执行时分支并不断重新运行电机控制功能的一个实例。

我曾尝试使用 concurrent.futures 导入，但收到消息说它不受支持，而且我找不到任何特定于我打算使用它的方式的用法。

Answer 1

我认为您不需要线程，但我可能会误解您的要求，因此我将提供 2 个备选方案。

1. 没有线程和 sleep

假设您当前的程序流程是这样的:

while True:
    data = read_sensors()
    command = process(data)
    motor_do(command)
    time.sleep(delay)

然后你可以取消 sleep ，并且只在最后一次调用至少 delay 秒前调用 motor_do。

last_call_time = -delay # to be sure that motor_do can be called the first time
# "On Windows, [time.clock] returns wall-clock seconds elapsed since the first call to this
# function, as a floating point number, based on the Win32 function QueryPerformanceCounter()
# The resolution is typically better than one microsecond." (python 2.7 doc)
# i.e. close to 0 on first call of time.clock()

while True:
    data = read_sensors()
    command = process(data)
    motor_try(command)

def motor_try(command):
    global last_call_time

    current_time = time.clock()
    # on win that works, on unix... you may want to take a look at the NB

    elapsed = current_time - last_call_time
    if elapsed >= delay:
        motor_do(command)
        last_call_time = current_time

1. 使用线程(这是一个例子，我没有使用 python 2.7 进行线程/异步的经验，所以可能有更好的方法来做到这一点)

假设您当前的程序流程是这样的:

while True:
    data = read_sensors()
    command = process(data)
    motor_do(command) // this function sleeps and you CANNOT change it

然后您必须启动 1 个线程，该线程只会将命令异步推送到电机。

import thread

command = None
command_lock = thread.allocate_lock()

def main():
    thread.start_new_thread(motor_thread)

    global command
    while True:
        data = read_sensors()
        with command_lock:
            command = process(data)

def motor_thread():
    while True:
        while not command: # just to be sure
            time.sleep(0.01)
            # small delay here (for instance, the time consumed by read_sensors + process)
        with command_lock:
            motor_do(command)
            command = None
        time.sleep(delay)

注意:在 Unix 上，time.clock() 返回处理器时间(= 没有空闲时间)，因此最好使用 time.time() ...除非更改系统时钟:“虽然此函数通常返回非递减值，但如果在两次调用之间已将系统时钟设置回原位，则它可以返回比先前调用更低的值。” ( python 2.7 文档)

我不知道 time.sleep() 如何对系统时钟变化使用react。

参见 How do I get monotonic time durations in python?对于 unix/py2.7 上的精确时间延迟(和 Understanding time.perf_counter() and time.process_time() 可能很有用)

Python3:只需使用 time.monotonic()... 或 time.perf_counter()。