5

从 stackoverflow 上的几篇文章中，我创建了这段代码。

设想

我想要一个 multiprocessing.queue 几个工人“听”

在键盘中断的情况下，主进程不应再将新项目放入队列中，并且在哨兵对象的帮助下，工作人员应该优雅地停止。

问题

我使用的当前版本的问题

signal.signal(signal.SIGINT, signal.SIG_IGN)

忽略 Ctrl + C 是它也被主进程忽略。

有任何想法吗 ？我需要使用多处理工作池吗？一些例子表明我可能不得不这样做。那么我还能使用队列吗？

from multiprocessing import Pool, Process,Queue
import time
import signal
# http://docs.python.org/3.1/library/multiprocessing.html#multiprocessing.Queue
# http://docs.python.org/3.1/library/multiprocessing.html#multiprocessing.Process


class Worker(Process):
    def __init__(self, queue,ident):
        super(Worker, self).__init__()
        # Ignore Signals
        signal.signal(signal.SIGINT, signal.SIG_IGN)
        self.queue= queue
        self.idstr= str(ident)
        print "Ident" + self.idstr

    def run(self):
        print 'Worker started'
        # do some initialization here

        print 'Computing things!'
        for data in iter( self.queue.get, None ):
            print "#" + self.idstr + " : " + str(data)
            time.sleep(5)
            print "#" + self.idstr + "Queue Size: " + str(self.queue.qsize())

        print "Worker Done"

#### Main ####
request_queue = Queue(10)

for i in range(4):
    Worker( request_queue,i ).start()

try:
    for data in range(1000000):
        request_queue.put( data )
        #print "Queue Size: " + str(request_queue.qsize())
        # Sentinel objects to allow clean shutdown: 1 per worker.
    for i in range(4):
        request_queue.put( None ) 

except KeyboardInterrupt:
    print "Caught KeyboardInterrupt, terminating workers"
    while  request_queue.empty()==False:
         request_queue.get()
    request_queue.put( None )
4

2 回答 2

1

根据您的解决方案（这很好），我添加了额外的保护层，以防主代码无响应并且用户取消了两次：

global STOP

import os, signal
def signal_handler(sig, frame):
    global STOP
    if STOP:
        signal.signal(signal.SIGINT, signal.SIG_IGN)
        os.kill(os.getpid(), signal.SIGTERM)
    STOP = True
signal.signal(signal.SIGINT, signal_handler)
于 2013-12-29T17:21:13.100 回答
0

我想我找到了解决办法。我仍然不喜欢我从 main 中获得 SIGINT 1 次，从 Worker 中获得 4 次，但也许我必须忍受这一点。

  1. 我为中断信号指定了一个信号处理程序。

  2. 收到第一个 Sig INT 后，我忽略更多 SIG Int 信号

  3. 我将停止标志切换为 TRUE

  4. 我打破了队列插入循环

  5. 我调用停止函数来清除队列并插入停止哨兵

    from multiprocessing import Pool, Process,Queue
import time
import signal
# http://docs.python.org/3.1/library/multiprocessing.html#multiprocessing.Queue
# http://docs.python.org/3.1/library/multiprocessing.html#multiprocessing.Process

# Stop Flag for loop
stop = False

# Define SIGINT
def signal_handler(sig, frame):
    print 'You pressed Ctrl+C!'
    global stop
    stop = True
    # Ignore more Ctrl+C
    signal.signal(signal.SIGINT, signal.SIG_IGN) 

signal.signal(signal.SIGINT, signal_handler)

def stopSentinel(request_queue):
    print "CTRL Stop Queue and insert None"

    # Empty Existing Queue
    while  request_queue.empty()==False:
        request_queue.get()

    # Put One None for each Worker
    for i in range(4):
        request_queue.put( None ) 


class Worker(Process):
    def __init__(self, queue,ident):
        super(Worker, self).__init__()

        self.queue= queue
        self.idstr= str(ident)
        print "Ident" + self.idstr

    def run(self):
        print 'Worker started'
        # do some initialization here

        print 'Computing things!'
        for data in iter( self.queue.get, None ):
            print "#" + self.idstr + " : " + str(data)
            time.sleep(5)
            print "#" + self.idstr + "Queue Size: " + str(self.queue.qsize())

        print "Worker Done"



#### Main #####
request_queue = Queue(10)

for i in range(4):
    Worker( request_queue, i ).start()

#### Fill Queue with Data ####
for data in range(1000000):
    request_queue.put( data )
    #print "Queue Size: " + str(request_queue.qsize())
    # Sentinel objects to allow clean shutdown: 1 per worker.

    # Check for Stop
    print "Check Breakout"
    if stop == True:
        print "Stop Break"
        break

if stop == True:
    stopSentinel(request_queue)
else:       
    print "Normal Stop" 
    for i in range(4):
        request_queue.put( None )
于 2013-06-26T11:54:10.883 回答