0

我有一个运行多个子进程的守护进程，这些子进程旨在维护 telnet 连接以从一堆气象站收集数据。我已经进行了设置，以便这些子进程永远从该 telnet 连接中读取数据，并通过multiprocessing.Queue. 当我停止守护进程时，我似乎无法让这些子进程干净地退出./test.py stop。有没有一种简单的方法可以在退出时关闭子进程？一个快速的谷歌提到有人使用multiprocessing.Event，在退出时设置此事件以确保进程退出的最佳方法是什么？这是我们当前的代码：

from daemon import runner
from multiprocessing import Process, Queue
import telnetlib

from django.utils.encoding import force_text

from observations.weather.models import WeatherStation
import os

os.environ['DJANGO_SETTINGS_MODULE'] = 'settings'


def read_weather_data(name, ip_address, port, queue):
    print "Started process to get data for", name
    client = telnetlib.Telnet(ip_address, port)

    while True:
        response = client.read_until('\r\n'.encode('utf8'))
        queue.put((name, force_text(response)))

    client.close()


class App(object):
    def __init__(self):
        self.stdin_path = '/dev/null'
        self.stdout_path = '/dev/tty'
        self.stderr_path = '/dev/tty'
        self.pidfile_path = '/tmp/process_weather.pid'
        self.pidfile_timeout = 5

    def run(self):
        queue = Queue()

        for station in WeatherStation.objects.filter(active=True):
            p = Process(target=read_weather_data,
                        args=(station.name, station.ip_address, station.port,
                              queue,))
            p.start()

        while True:
            name, data = queue.get()
            print "Received data from ", name
            print data

app = App()
daemon_runner = runner.DaemonRunner(app)
daemon_runner.do_action()
4

1 回答 1

2

似乎已经找到了一种方法来做到这一点，但我不确定这是否是最好的方法。

from daemon import runner
from multiprocessing import Process, Queue, Event
import telnetlib

from django.utils.encoding import force_text

from observations.weather.models import WeatherStation
import os
import signal
import errno

os.environ['DJANGO_SETTINGS_MODULE'] = 'settings'


def read_weather_data(name, ip_address, port, queue, exit):
    print "Started process to get data for", name
    client = telnetlib.Telnet(ip_address, port)

    while not exit.is_set():
        response = client.read_until('\r\n'.encode('utf8'))
        queue.put((name, force_text(response)))

    print "exit called for", name
    client.close()


def exit_handler(signum, frame):
    print "exiting..."


class App(object):
    def __init__(self):
        self.stdin_path = '/dev/null'
        self.stdout_path = '/dev/tty'
        self.stderr_path = '/dev/tty'
        self.pidfile_path = '/tmp/process_weather.pid'
        self.pidfile_timeout = 5

    def run(self):
        exit = Event()

        def exit_handler(signum, frame):
            print "exiting..."
            exit.set()

        signal.signal(signal.SIGTERM, exit_handler)
        queue = Queue()

        workers = []
        for station in WeatherStation.objects.filter(active=True):
            p = Process(target=read_weather_data,
                        args=(station.name, station.ip_address, station.port,
                              queue, exit))
            workers.append(p)

        for worker in workers:
            worker.start()

        while True:
            try:
                name, data = queue.get()
            except IOError as e:
                # we received a signal whilst waiting for I/O
                if e.errno != errno.EINTR:
                    raise
                else:
                    break
            print "Received data from ", name
            print data

        for worker in workers:
            worker.join()


app = App()
daemon_runner = runner.DaemonRunner(app)
daemon_runner.do_action()
于 2013-08-21T07:49:13.610 回答