最近在項(xiàng)目中調(diào)用第三方接口時(shí)候，經(jīng)常會(huì)出現(xiàn)請(qǐng)求超時(shí)的情況，或者參數(shù)的問(wèn)題導(dǎo)致調(diào)用異代碼異常。針對(duì)超時(shí)異常，查詢了python 相關(guān)文檔，沒(méi)有并發(fā)現(xiàn)完善的包來(lái)根據(jù)用戶自定義的時(shí)間來(lái)拋出超時(shí)異常的模塊。所以自己干脆自己來(lái)實(shí)現(xiàn)一個(gè)自定義的超時(shí)異常。目前找到了兩種方式來(lái)實(shí)現(xiàn)超時(shí)異常的功能（signal.alarm()、threading實(shí)現(xiàn)超時(shí)異常）

方法1 thread + time　

原理：將要調(diào)用的功能函數(shù)放入子線程，通過(guò)設(shè)定子線程的阻塞時(shí)間，超時(shí)則主線程并不會(huì)等待子線程的執(zhí)行。主線程退出，子線程就不存在了。

核心就是在程序中添加 join()方法，用于等待線程結(jié)束。join()的作用是，在子線程完成運(yùn)行之前，這個(gè)子線程的父線程將會(huì)被一直阻塞.

# coding=utf-8
import threading
import time


def myFunc():
  time.sleep(4)
  print("myFunc執(zhí)行了")


if __name__ == '__main__':
  t = threading.Thread(target=myFunc)
  t.setDaemon(True)
  t.start()

  t.join(2)
  print("it's over")

執(zhí)行結(jié)果：

it's over

可以看出，當(dāng)主線程執(zhí)行到2秒時(shí)候，結(jié)束退出。子線程還沒(méi)有結(jié)束，沒(méi)有執(zhí)行完及被強(qiáng)制退出

# coding=utf-8
import threading
import time


def myFunc():
  time.sleep(1)
  print("myFunc執(zhí)行了")


if __name__ == '__main__':
  t = threading.Thread(target=myFunc)
  t.setDaemon(True)
  t.start()

  t.join(2)
  print("it's over")

顯示結(jié)果：

myFunc執(zhí)行了
it's over

可以看出，子線程結(jié)束時(shí)，用時(shí)1秒，沒(méi)有超過(guò)主線程設(shè)定的3秒，所以主線程與子線程都被執(zhí)行了

方法 2  signal.alarm() ，注意兩點(diǎn)：一是signal信號(hào)機(jī)制要在linux上才能運(yùn)行； 二是signal信號(hào)在主線程中才會(huì)會(huì)起作用

import signal
import time

# Define signal handler function
def myHandler(signum, frame):
  exit("TimeoutError")

def test_fun():
  # time.sleep(3)
  int("afsdf")
  a = 2 + 3
  return a

if __name__ == '__main__':
  try:
    signal.signal(signal.SIGALRM, myHandler)
    signal.alarm(2)
    test = test_fun()
    print(test)
    signal.alarm(0)
  except Exception as ret:
    print("msg:", ret)

執(zhí)行結(jié)果：

當(dāng) time.sleep(3) 時(shí)，會(huì)拋出TimeoutError的異常
當(dāng) test_fun 里面出現(xiàn) int("afsdf")時(shí)， 會(huì)拋出 ValueError("invalid literal for int()         with base 10: 'afsdf'",))
當(dāng)test_fun函數(shù)執(zhí)行的時(shí)間小于2 秒時(shí)，就會(huì)返回函數(shù)對(duì)應(yīng)的值

方法3  帶有返回值的超時(shí)異常，可以通過(guò)創(chuàng)建thread類的方式來(lái)進(jìn)行捕捉

import threading
import sys
import time


class Dispacher(threading.Thread):
  def __init__(self, fun, args):
    threading.Thread.__init__(self)
    self.setDaemon(True)
    self.result = None
    self.error = None
    self.fun = fun
    self.args = args

    self.start()

  def run(self):
    try:
      self.result = self.fun(self.args)
    except:
      self.error = sys.exc_info()


def test_fun(i):
  # time.sleep(4)
  a = i*i
  # b 　　　
　　return a
def main_fun():
  c = Dispacher(test_fun, 2)
  c.join(2)

  if c.isAlive():
    return "TimeOutError"
  elif c.error:
    return c.error[1]
  t = c.result
  return t

if __name__ == '__main__':
  fun = main_fun()
  print(fun)

顯示結(jié)果：

test_fun 執(zhí)行時(shí)間大于設(shè)置的2秒時(shí)，會(huì)拋出TimeOutError
test_fun 執(zhí)行時(shí)間小于設(shè)置的2秒時(shí)，并且函數(shù)正常執(zhí)行時(shí)，顯示：4
test_fun 里面出現(xiàn)比如 “b” 時(shí)，會(huì)拋出 global name 'b' is not defined 的異常

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。