60秒后结束python代码

Question

下面是一些功能齐全的代码。

我计划通过命令行执行此代码，但我希望它在 60 秒后结束。

有谁知道最好的解决方法？

提前致谢。

import time
class listener(StreamListener):

    def on_data(self, data):
        try:
            print data
            saveFile = open('twitDB.csv','a')
            saveFile.write(data)
            saveFile.write('\n')
            saveFile.close()
            return True
        except BaseException, e:
            print 'failed ondata,' ,str(e)
            time.sleep(5)

    def on_error(self, status):
        print status

Answer 1

您可以将您的代码移动到一个守护线程并在 60 秒后退出主线程：

#!/usr/bin/env python
import time
import threading

def listen():
    print("put your code here")

t = threading.Thread(target=listen)
t.daemon = True
t.start()

time.sleep(60)
# main thread exits here. Daemon threads do not survive.

Answer 2

试试这个：

import os
import time
from datetime import datetime
from threading import Timer

def exitfunc():
    print "Exit Time", datetime.now()
    os._exit(0)

Timer(5, exitfunc).start() # exit in 5 seconds

while True: # infinite loop, replace it with your code that you want to interrupt
    print "Current Time", datetime.now()
    time.sleep(1)

这个 StackOverflow 问题中还有更多示例：Executing periodic actions in Python

我认为不鼓励使用os._exit(0)，但我不确定。关于这件事的一些事情感觉不洁。不过，它确实有效。

Answer 3

我希望这是一种定期执行函数并在 60 秒后结束的简单方法：

import time
import os
i = 0
def executeSomething():
 global i
 print(i)
 i += 1
 time.sleep(1)
 if i == 10:
    print('End')
    os._exit(0)


while True:
 executeSomething()

Answer 4

使用signal.ALARM 在指定时间后收到通知。

import signal, os

def handler(signum, frame):
    print '60 seconds passed, exiting'
    cleanup_and_exit_your_code()

# Set the signal handler and a 60-second alarm
signal.signal(signal.SIGALRM, handler)
signal.alarm(60)

run_your_code()

从您的示例来看，代码究竟会做什么、如何运行以及它将迭代什么样的循环并不明显。但是您可以轻松实现ALARM 信号，以便在超时到期后得到通知。

Answer 5

这是我最喜欢的超时方式。

def timeout(func, args=None, kwargs=None, TIMEOUT=10, default=None, err=.05):
    if args is None:
        args = []
    elif hasattr(args, "__iter__") and not isinstance(args, basestring):
        args = args
    else:
        args = [args]

    kwargs = {} if kwargs is None else kwargs

    import threading
    class InterruptableThread(threading.Thread):
        def __init__(self):
            threading.Thread.__init__(self)
            self.result = None

        def run(self):
            try:
                self.result = func(*args, **kwargs)
            except:
                self.result = default

    it = InterruptableThread()
    it.start()
    it.join(TIMEOUT* (1 + err))
    if it.isAlive():
        return default
    else:
        return it.result