如何运行其函数可以作为线程调用的对象

Question

我正在用 Python 为 Raspberry Pi 编写脚本，其目的是监听服务器/消息代理的命令并使用特定硬件执行所述命令。有时，这些命令必须持续指定的持续时间（即我需要打开某些东西，保持 t 秒，然后关闭），这是通过让代码在 on 和 off 命令之间的所述持续时间内休眠来实现的（这发生在内部一个函数调用——hardware1.on(dur = t))。我希望能够用另一个命令中断该序列（例如在 t 秒结束之前关闭硬件）。我尝试通过多处理来完成此操作，但无法获得我正在寻找的行为。

此硬件（不同颜色的灯杆）通过 LiteStalk 类进行控制。此类由 Lite 对象（茎中的每个灯）组成，它们也有自己的类。这两个类都继承了 multiprocessing.process。在创建特定 LiteStalk 然后侦听消息代理（基于 MQTT）以获取命令的主要代码中，我评估发布到代理的命令（这是在消息发布到代理时运行的 on_message 回调） .

import time
import LiteCntrlModule as LiteStalkMod
import multiprocessing
import paho.mqtt.client as mqtt

print('Starting...\n')

# Set gpio designatin mode to BCM
gpio.setmode(gpio.BCM)

# Initialize light stalk
stalkdict = {'red':1, 'yel':2, 'grn':3, 'bzr':4}
stalk = LiteStalkMod.LiteStalk(stalkdict)
msgRec = ""

def on_connect(client, userdata, flags, rc):
    print("Connected with result code "+str(rc))
    if(rc == 0):
        print('Code "0" indicates successful connection.  Waiting for messages...')
    # Subscribing in on_connect() means that if we lose the connection and
    # reconnect then subscriptions will be renewed.
    client.subscribe("asset/andon1/state")

# The callback for when a PUBLISH message is received from the server.
def on_message(client, userdata, msg):
    print(msg.topic+" "+str(msg.payload))
    msgRec = msg.payload
    eval(msg.payload)
    if msg.payload == "stalk.off()":
        print("If this executes while another command is running, it works!")

client = mqtt.Client(client_id="")
client.username_pw_set("mytopic", password="mypassword")
client.on_connect = on_connect
client.on_message = on_message

client.connect("mymessagebrokeraddress", 1883, 60)
client.subscribe("mytopic")

# Blocking call that processes network traffic, dispatches callbacks and
# handles reconnecting.
# Other loop*() functions are available that give a threaded interface and a
# manual interface.

try:
    client.loop_start() # start listening in a thread and proceed

except KeyboardInterrupt: # so that aborting with Ctrl+C works cleanly
stalk.off()

finally:
stalk.shutDown()

LiteCtnrlModule（Lite 和 LiteStalk 类）如下：

import time
import multiprocessing
from relay_lib_seeed import *

class Lite(multiprocessing.Process)：

# A Lite object has an associated relay and functions
# Ex: red
    # A lite can be controlled
    # Ex: red.blink()       

def __init__(self, relayIn):
    # Ex: red = Lite.Lite(1)
    multiprocessing.Process.__init__(self) # allows you to create multiple objects that can be run as threads
    self.daemon = True # creates a daemon thread that will exit when the main code terminates
    self.start() # allows multiproc. to begin
    self.relay = relayIn

def shutDown(self):
    # terminates the threaded object
    relay_off(self.relay)
    self.join()

def off(self, dur = 0):
            # turns light off

def on(self, dur = 0):
            # turns light on, optional duration to stay on for

# blink
def blink(self, dur = 0, timeOn = .5, timeOff = .5):
            # blinks light

类 LiteStalk(multiprocessing.Process)：

# A LiteStalk object can have any number of "lite" onjects in it. Ex:
    # Object: stalk1
        # A lite object in stalk1 respresents one segment/color of the light stalk
        # stalk1.red
            # Any lite can be turned on/off in various patterns for amounts of time, etc.
            # stalk1.red.blink()
        # An entire stalk can be controlled all at once
        # stalk1.cycle()

liteList = {}

def __init__(self, liteListIn):

    # liteListIn = {'clr1':relay1, 'clr2":relay2, 'clr3':relay3]...}
    self.liteList = liteListIn;
    multiprocessing.Process.__init__(self) # allows you to create multiple objects that can be run as threads
    self.daemon = True # creates a daemon thread that will exit when the main code terminates
    self.start() # allows multiproc. to begin
    for lite in self.liteList: # for each lite color string in the lites dict
        setattr(self, lite, Lite(self.liteList[lite])) # creates a lite obj attr in the LiteStalk obj
    print(self.liteList)

def shutDown(self):

    # each light is turned off and that gpio pin is cleaned-up
    relay_all_off()
    self.join() # joins thread

def off(self, dur = 0):
            # turns all hardware off

def on(self): 
            # turns all hardware on, optional duration to stay on for
def blink(self, timeOn, timeOff):
            # blinks all hardware

def cntDn(self, dur = 20, yelDur = 2, redDur = 10): #in min
    # enters a count down sequence

在执行发布到服务器的任何其他命令之前，该命令始终运行到完成，即，茎在命令的持续时间内保持打开状态，并且在持续时间结束之前不能被命令关闭（或其他任何东西）。我认为这可能是因为我没有在 run() 函数中包含我的多处理对象的所有功能，但我没有运气就搞砸了。

Answer 1

Paho MQTT 客户端是单线程的（您以 client.loop_start() 函数开始的线程）它一次只能为一条消息调用 on_message()。

这意味着它将阻塞对eval() 的调用，直到传递给它的所有内容完成，即使那是代码正在创建新线程来做事。

Answer 2

我将建议用 threading.Event 或等效项上的超时来替换睡眠，然后检查睡眠何时结束，是否是由于设置了事件或超时。如果设置了事件，请停止。

但似乎除了可中断的睡眠之外还有其他问题。