如何在线程化Python程序中捕获SIGINT？

发布于 2024-09-25 19:44:38 字数 469 浏览 4 评论 0原文

当使用模块线程和类 Thread() 时，我无法在控制台中捕获 SIGINT (Ctrl + C ) 无法被捕获。

为什么以及我能做什么？

简单的测试程序：

#!/usr/bin/env python

import threading

def test(suffix):
    while True:
        print "test", suffix

def main():
    for i in (1, 2, 3, 4, 5):
        threading.Thread(target=test, args=(i, )).start()

if __name__ == "__main__":
    main()

当我点击 Ctrl + C 时，没有任何反应。

原文

When using module threading and class Thread(), I can't catch SIGINT (Ctrl + C in console) can not be caught.

Why and what can I do?

Simple test program:

#!/usr/bin/env python

import threading

def test(suffix):
    while True:
        print "test", suffix

def main():
    for i in (1, 2, 3, 4, 5):
        threading.Thread(target=test, args=(i, )).start()

if __name__ == "__main__":
    main()

When I hit Ctrl + C, nothing happens.

分享到QQ

分享到微博

如果你对这篇内容有疑问，欢迎到本站社区发帖提问参与讨论，获取更多帮助，或者扫码二维码加入 Web 技术交流群。

发布评论

需要登录才能够评论，你可以免费注册一个本站的账号。

紙鸢 2024-10-02 19:44:38

线程和信号不能混合。在 Python 中，情况比在外部更是如此：信号只能传递到一个线程（主线程）；其他线程不会收到该消息。除了主线程之外，您无法中断其他线程。他们超出了你的控制范围。

您在这里唯一能做的就是使用 queue 模块在主线程和您启动的任何线程之间引入一个通信通道。然后，您可以向线程发送一条消息，并让它在看到该消息时终止（或执行您想要的任何其他操作）。

或者，这通常是一个非常好的选择，就是不使用线程。然而，使用什么在很大程度上取决于您想要实现的目标。

回复收藏 0 原文

蓝色星空 2024-10-02 19:44:38

基本上，您可以通过在工作期间读取队列来检查父级是否发出信号。如果父进程收到 SIGINT，那么它会在队列中发出一个信号（在本例中为任何信号），子进程结束他们的工作并退出......

def fun(arg1, thread_no, queue):
   while True:
    WpORK...
    if queue.empty() is False or errors == 0:
     print('thread ', thread_no, ' exiting...')
     with open('output_%i' % thread_no, 'w') as f:
      for line in lines: f.write(line)
     exit()

threads = []
for i, item in enumerate(items):
 threads.append( dict() )
 q = queue.Queue()
 threads[i]['queue'] = q
 threads[i]['thread'] = threading.Thread(target=fun, args=(arg1, i, q))
 threads[i]['thread'].start()
try:
 time.sleep(10000)
except:
 for thread in threads:
  thread['queue'].put('TERMINATING')

Basically you can check if the parent issued a signal by reading a queue during the work. If the parent receives a SIGINT then it issues a signal across the queue (in this case anything) and the children wrap up their work and exit...

def fun(arg1, thread_no, queue):
   while True:
    WpORK...
    if queue.empty() is False or errors == 0:
     print('thread ', thread_no, ' exiting...')
     with open('output_%i' % thread_no, 'w') as f:
      for line in lines: f.write(line)
     exit()

threads = []
for i, item in enumerate(items):
 threads.append( dict() )
 q = queue.Queue()
 threads[i]['queue'] = q
 threads[i]['thread'] = threading.Thread(target=fun, args=(arg1, i, q))
 threads[i]['thread'].start()
try:
 time.sleep(10000)
except:
 for thread in threads:
  thread['queue'].put('TERMINATING')

回复收藏 0 原文

~没有更多了~