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

Question

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

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

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

现在我需要用它制作一个合适的 linux 守护进程（可以是 start、stop 和 restart 通过 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 的新实例是在下面创建的，并且在不同的进程中我们有不同的队列（以及不同的 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)