python - 创建使用 Multiprocessing 和 Multiprocessing.Queues 的 linux 守护进程

Question

我的任务是监听 UDP 数据报，对其进行解码(数据报具有二进制信息)，将解码后的信息放入字典中，将字典转储为 json 字符串，然后将 json 字符串发送到远程服务器(ActiveMQ)。

解码和发送到远程都可能很耗时。为了使程序更具可扩展性，我们创建了两个进程(Multiprocessing.Process):

• Listner(监听数据报，分析，创建json并放入Multiprocessing.Queue)
• 发件人(不断尝试从队列中获取 json 字符串到数组，如果数组长度超过阈值 - 将所有收集到的字符串发送到远程服务器)

现在我需要从它创建一个合适的 linux 守护进程(可以通过 service 命令启动、停止和重新启动)。

问题:如何从 python 多处理程序创建守护进程。我没有找到关于此的指南。有没有人知道如何做到这一点，或者有工作示例。

---

以下文字是我为实现这一目标所做的努力: 我找到了 python 守护进程的小例子:http://www.gavinj.net/2012/06/building-python-daemon-process.html 所以我重写了我的代码(抱歉代码太长):

import socket
import time
import os    
from select import select    
import multiprocessing
from multiprocessing import Process, Queue, Value

import stomp
import json

import logging
logger = logging.getLogger("DaemonLog")
logger.setLevel(logging.INFO)
formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
handler = logging.FileHandler("/var/log/testdaemon/testdaemon.log")
handler.setFormatter(formatter)
logger.addHandler(handler)
log = logger
#Config listner
domain = 'example.host.ru'
port = int(9930)

#Config remote queue access
queue_cfg = {
    'host': 'queue.test.ru',
    'port': 61113,
    'user': 'user',
    'password': 'pass',
    'queue': '/topic/test.queue'
}

class UDPListener():
    def __init__(self, domain, port, queue_cfg):
        # If I initialize socket during init I see strange error:
        # on the line: data, addr = sock_inst.recvfrom(int(10000))
        # error: [Errno 88] Socket operation on non-socket
        # So I put initialization to runListner function
        #self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        #self.sock.bind((domain, port))
        self.domain = domain
        self.port = port
        self.remote_queue_cfg = queue_cfg
        self.queue = Queue()
        self.isWorking = Value('b', True)
        self.decoder = Decoder()
        self.reactor = ParallelQueueReactor(self.queue)

        self.stdin_path = '/dev/null'
        self.stdout_path = '/dev/tty'
        self.stderr_path = '/dev/tty'
        self.pidfile_path = '/var/run/testdaemon/testdaemon.pid'
        self.pidfile_timeout = 5

    def __assignData(self, addr, data):
        receive_time = time.time()
        messages = self.decoder.decode(receive_time, addr, data)
        for msg in messages:
            self.reactor.addMessage(msg)

    def runListner(self):
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.sock.bind((domain, port))
        while self.isWorking.value:
            inputready, outputready, exceptready = select([self.sock], [], [])
            for sock_inst in inputready:
                if sock_inst == self.sock:
                    data, addr = sock_inst.recvfrom(int(10000))
                if data:
                    self.__assignData(addr[0], data)
        self.sock.close()

    def runQueueDispatcher(self):
        while self.isWorking.value:
            connected = False
            while not connected:
                try:
                    conn = stomp.Connection(host_and_ports=[(self.remote_queue_cfg['host'], self.remote_queue_cfg['port'])])
                    conn.start()
                    conn.connect(self.remote_queue_cfg['user'], self.remote_queue_cfg['password'], wait=True)
                    connected = True
                except socket.error:
                    log.error('Could not connect to activemq server.')
                    time.sleep(20)

                if connected == True:
                    while self.isWorking.value:
                        msg = None
                        if not self.queue.empty():
                            #Now error appear hear even when not self.queue.empty()
                            msg = self.queue.get()
                        else:
                            time.sleep(1)

                        if msg is not None:
                            try:
                                data = json.dumps(msg)
                                conn.send(body=data, destination=self.remote_queue_cfg['queue'])
                                count += 1
                            except:
                                log.error('Failed to send message to queue.')
                                time.sleep(1)

    def stop(self):
        self.isWorking.value = False

    def run(self):
        log.error('StartProcesses')
        dispatcher_process = Process(target=self.runQueueDispatcher, name='Dispatcher')
        listner_process = Process(target=self.runListner, name='Listner')
        dispatcher_process.start()
        listner_process.start()
        dispatcher_process.join()
        listner_process.join()
        log.info('Finished')
#------------------------------------------------------------------
def main():
    from daemon import runner

    app = UDPListener(domain, port, queue_cfg)

    daemon_runner = runner.DaemonRunner(app)
    daemon_runner.daemon_context.files_preserve=[handler.stream]
    daemon_runner.do_action()

if __name__ == "__main__":
        main()

现在我在 msg = self.queue.get() 上看到错误

Traceback (most recent call last):   File "/usr/lib64/python2.6/multiprocessing/process.py", line 232, in
_bootstrap
    self.run()   File "/usr/lib64/python2.6/multiprocessing/process.py", line 88, in run
    self._target(*self._args, **self._kwargs)   File "/root/ipelevan/dream/src/parallel_main.py", line 116, in runQueueDispatcher
    msg = self.queue.get()   File "/usr/lib64/python2.6/multiprocessing/queues.py", line 91, in get
    res = self._recv() EOFError

我在手动运行 UDPListner.run() 时没有看到这个错误。但是对于 daemon runner，它看起来像是在下面创建了 UDPListner 的新实例，并且在不同的进程中我们有不同的 Queue(以及不同的 self.socket，当它在 init 中初始化时)。

Answer 1

首先:将共享对象(队列、值)的链接保留为类的成员以供进程使用是一个坏主意。它以某种方式在没有妖魔化的情况下起作用。但是，当在 DaemonContext 中运行相同的代码时，os.fork() 发生了，并以某种方式弄乱了对象的链接。我不太确定 Multiprocessing 模块是否设计为在对象的方法内 100% 正确工作。

其次:DaemonContext 有助于从 shell 中分离进程、重定向流并执行与守护进程相关的其他几项操作，但我还没有找到任何好的方法来检查这样的守护进程是否已经在运行。所以我只是用了

if os.path.isfile(pidfile_path):
        print 'pidfile %s exists. Already running?' % pidfile_path
        sys.exit(1)