python中自定義超時異常的幾種方法

　　最近在項目中調用第三方接口時候，常常會出現請求超時的狀況，或者參數的問題致使調用異代碼異常。針對超時異常，查詢了python 相關文檔，沒有並發現完善的包來根據用戶自定義
的時間來拋出超時異常的模塊。因此本身乾脆本身來實現一個自定義的超時異常。目前找到了兩種方式來實現超時異常的功能（signal.alarm()、threading實現超時異常）
方法1 thread + time　

原理：將要調用的功能函數放入子線程，經過設定子線程的阻塞時間，超時則主線程並不會等待子線程的執行。主線程退出，子線程就不存在了。
核心就是在程序中添加 join()方法，用於等待線程結束。join()的做用是，在子線程完成運行以前，這個子線程的父線程將會被一直阻塞.

 1 # coding=utf-8
 2 import threading
 3 import time
 4 
 5 
 6 def myFunc():
 7     time.sleep(4)
 8     print("myFunc執行了")
 9 
10 
11 if __name__ == '__main__':
12     t = threading.Thread(target=myFunc)
13     t.setDaemon(True)
14     t.start()
15 
16     t.join(2)
17     print("it's over")

執行結果：

　　it's over

能夠看出，當主線程執行到2秒時候，結束退出。子線程尚未結束，沒有執行完及被強制退出

 1 # coding=utf-8
 2 import threading
 3 import time
 4 
 5 
 6 def myFunc():
 7     time.sleep(1)
 8     print("myFunc執行了")
 9 
10 
11 if __name__ == '__main__':
12     t = threading.Thread(target=myFunc)
13     t.setDaemon(True)
14     t.start()
15 
16     t.join(2)
17     print("it's over")
顯示結果：

　　myFunc執行了
　　it's over

能夠看出，子線程結束時，用時1秒，沒有超過主線程設定的3秒，因此主線程與子線程都被執行了

方法 2  signal.alarm() ，注意兩點：一是signal信號機制要在linux上才能運行； 二是signal信號在主線程中才會會起做用

 1 import signal
 2 import time
 3 
 4 
 5 # Define signal handler function
 6 def myHandler(signum, frame):
 7     exit("TimeoutError")
 8 
 9 
10 def test_fun():
11     # time.sleep(3)
12     int("afsdf")
13     a = 2 + 3
14 
15     return a
16 
17 
18 
19 if __name__ == '__main__':
20     try:
21         signal.signal(signal.SIGALRM, myHandler)
22         signal.alarm(2)
23         test = test_fun()
24         print(test)
25         signal.alarm(0)
26     except Exception as ret:
27         print("msg:", ret)

執行結果：
　　當 time.sleep(3) 時，會拋出TimeoutError的異常
　　當 test_fun 裏面出現 int("afsdf")時， 會拋出 ValueError("invalid literal for int() with base 10: 'afsdf'",))
　　當test_fun函數執行的時間小於2 秒時，就會返回函數對應的值

方法3  帶有返回值的超時異常，能夠經過建立thread類的方式來進行捕捉

 1 import threading
 2 import sys
 3 import time
 4 
 5 
 6 class Dispacher(threading.Thread):
 7     def __init__(self, fun, args):
 8         threading.Thread.__init__(self)
 9         self.setDaemon(True)
10         self.result = None
11         self.error = None
12         self.fun = fun
13         self.args = args
14 
15         self.start()
16 
17     def run(self):
18         try:
19             self.result = self.fun(self.args)
20         except:
21             self.error = sys.exc_info()
22 
23 
24 def test_fun(i):
25     # time.sleep(4)
26     a = i*i
27     # b 　　　
29 　　return a
30 def main_fun():
31     c = Dispacher(test_fun, 2)
32     c.join(2)
33 
34     if c.isAlive():
35         return "TimeOutError"
36     elif c.error:
37         return c.error[1]
38     t = c.result
39     return t
40 
41 if __name__ == '__main__':
42     fun = main_fun()
43     print(fun)
     
顯示結果：

　　test_fun 執行時間大於設置的2秒時，會拋出TimeOutError　　test_fun 執行時間小於設置的2秒時，而且函數正常執行時，顯示：4　　test_fun 裏面出現好比 「b」 時，會拋出 global name 'b' is not defined 的異常