python中的正确线程

Question

我正在编写一个家庭自动化助手——它们基本上是类似守护进程的小型 Python 应用程序。他们可以将每个进程作为一个单独的进程运行，但既然会这样，我决定我将建立一个小型调度程序，它将在它们自己的线程中生成每个守护进程，并且能够在将来线程死亡时采取行动。

这就是它的样子（使用两个类）：

from daemons import mosquitto_daemon, gtalk_daemon
from threading import Thread

print('Starting daemons')
mq_client = mosquitto_daemon.Client()
gt_client = gtalk_daemon.Client()

print('Starting MQ')
mq = Thread(target=mq_client.run)
mq.start()

print('Starting GT')
gt = Thread(target=gt_client.run)
gt.start()

while mq.isAlive() and gt.isAlive():
    pass
print('something died')

问题是 MQ 守护进程 (moquitto) 可以正常工作，我是否应该直接运行它：

mq_client = mosquitto_daemon.Client()
mq_client.run()

它将启动并挂在那里收听所有涉及相关主题的消息 - 正是我正在寻找的。​​p>

但是，在调度程序中运行会使其行为怪异 - 它会收到一条消息然后停止执行，但报告线程仍处于活动状态。鉴于它在没有线程woodoo 的情况下工作正常，我假设我在调度程序中做错了什么。

我引用 MQ 客户端代码以防万一：

import mosquitto
import config
import sys
import logging


class Client():
    mc = None

    def __init__(self):
        logging.basicConfig(format=u'%(filename)s:%(lineno)d %(levelname)-8s [%(asctime)s]  %(message)s', level=logging.DEBUG)
        logging.debug('Class initialization...')
        if not Client.mc:
            logging.info('Creating an instance of MQ client...')
            try:
                Client.mc = mosquitto.Mosquitto(config.DEVICE_NAME)
                Client.mc.connect(host=config.MQ_BROKER_ADDRESS)
                logging.debug('Successfully created MQ client...')
                logging.debug('Subscribing to topics...')
                for topic in config.MQ_TOPICS:
                    result, some_number = Client.mc.subscribe(topic, 0)
                    if result == 0:
                        logging.debug('Subscription to topic "%s" successful' % topic)
                    else:
                        logging.error('Failed to subscribe to topic "%s": %s' % (topic, result))
                logging.debug('Settings up callbacks...')
                self.mc.on_message = self.on_message
                logging.info('Finished initialization')
            except Exception as e:
                logging.critical('Failed to complete creating MQ client: %s' % e.message)
                self.mc = None
        else:
            logging.critical('Instance of MQ Client exists - passing...')
            sys.exit(status=1)

    def run(self):
        self.mc.loop_forever()

    def on_message(self, mosq, obj, msg):
        print('meesage!!111')
        logging.info('Message received on topic %s: %s' % (msg.topic, msg.payload))

Answer 1

你正在传递 Thread 另一个类实例的 run 方法...它真的不知道如何处理它。

threading.Thread 可以以两种一般方式使用：生成一个 Thread 包装的独立函数，或者作为具有 run 方法的类的基类。 在您的情况下，似乎基类是要走的路，因为您的 Client 类有一个 run 方法。

在您的 MQ 类中替换以下内容，它应该可以工作：

from threading import Thread

class Client(Thread):
    mc = None

    def __init__(self):
        Thread.__init__(self) # initialize the Thread instance
        ...
    ...

    def stop(self):
        # some sort of command to stop mc
        self.mc.stop() # not sure what the actual command is, if one exists at all...

那么调用的时候，不用Thread：

mq_client = mosquitto_daemon.Client()
mq_client.start() 

print 'Print this line to be sure we get here after starting the thread loop...'

Answer 2

需要考虑的几件事：

1. zeromq 讨厌在 1 个线程中初始化并在另一个线程中运行。您可以按照建议将 Client() 重写为 Thread，或者编写自己的函数来创建 Client 并在线程中运行该函数。

2. Client() 有一个类级别变量 mc。我假设 mosquitto_daemon 和 gtalk_daemon 都使用相同的客户端，因此它们在争夺哪个 Client.mc 获胜。

3. "while mq.isAlive() and gt.isAlive(): pass" 会吃掉整个处理器，因为它只是不停地轮询而没有睡眠。考虑到 python 只是准线程的（全局解释器锁 (GIL) 一次只允许运行 1 个线程），这将使您的“守护进程”停滞不前。

4. 同样考虑到 GIL，原来的守护进程实现可能会更好。