python多线程愚见-附重启脚本

基础

python多线程由于底层有GIL,实际上是单线程。因此把一件事或很多件事用python多线程完成,比直接调用可能还要慢(个人感受,如进行密码爆破,后台爆破),直接迭代操作比用多线程逐个传参更快。但python的多线程并不是鸡肋,我更多地把它当成“后台运行”,即用多线程完成一些不需要即使获得返回值,让主程序在这个点上可以继续前进。

简单使用

python有线程池和多线程,线程池是单个线程拿着池中一个参数进行操作,完成后返回,再拿着池中另一个参数继续执行操作,它们互不干扰独立分工。我在我的压缩包爆破、简易扫描器中也是用这种方法。

from concurrent.futures import ThreadPoolExcutor
argvs = []   # 把参数记录入list中
with ThreadPoolExcutor(max_workers=500) as pool:   # 设置最大线程数
    results = pool.map(FuncName,argvs)  # 这种方法是传入数组参数当参数池
for res in results:
    print(res)  # 逐个打印结果,即使线程没跑完还是能逐个打印

但上面这个方法我并不喜欢,实际上它是一次执行max_workers个线程异步执行调用,完成后再返回结果。与我的“后台运行”想法并不符合。所以我还是喜欢用threading

import threading
thread1 = threading.Thread(target=FuncName,args=(argv1,argv2,...))  # 简单声明
thread1.start()

threading应用场景(部分)

以我曾经写过的小工具为例子:
1.后台挂起、无返回值
以编写一个调用sqlmapapi接口为例,先说表现,我只需要把url等参数传进sqlmapapi.py即可开启扫描。但一开始先要运行sqlmapapi.py,如果每次都手动开启sqlmapapi.py的话,就不符合工具自动化的理念,但如果只是普通地使用os.system()subprocess在脚本一开始调用sqlmapapi.py,则程序永远都不会运行下去。因为调用这些函数并不会获得返回值,调用sqlmapapi.py,只会一直监听。
这种情况,就可以用多线程解决,因为主程序无需等待多线程得到返回值也可继续运行下去:

import threading

def start_api():
    cmd = 'python sqlmapapi.py -s'
    p = os.system(cmd)

thread1 = threading.Thread(target=start_api)
thread1.start()

2.耗时任务
一些不过于影响主程序,比较耗时的任务可以用多线程来解决。如图片文件等上传、单个网页扫描等。

3.加强程序稳定性
最近帮人写了一个量化交易监测脚本,这东西最重要的是稳定性。但使用单一线程(即不用多线程)运行时,遇到一些网络波动,api拒绝连接等外部因素,那是得写多少个try-except才能解决。即使某一步加入了异常处理模块,但后续操作需要的参数值就对应不上了,还是会运行出错。这种牵一发动全身的,最好的方法是“重启程序”。但python不能重启自身,但可以“重启”线程。

首先把主程序分成几部分用多线程运行:
– 开仓监测
– 平仓监测
– api连接监测
– 监测上述三个线程执行情况

分别以thread1thread2thread3命名上三个线程,最后一个线程是整个程序执行的关键,必须保证不会崩溃(实际上写好了就不会无缘无故崩)。

现在来写一下如何“重启”多线程

python多线程程序运行中,会出现由于异常而导致某线程停止的情况,为了保证程序的稳定运行,需要自动重启down掉的线程。可以把线程当成一个独立的程序,而且A线程崩了不会影响主程序及其它线程(除非有参数交集)。在网上看的重启线程都不符合我的要求,他们写的大多数是重启一些类似执行线程池任务的,而我这种是重启一个个独立线程的。
但我稍微改写了一下,贴上thread4代码:

import threading,time

def reStartT(t1,t2,t3,arg1,arg2):   # 根据需要把参数传进来,t线程,a参数,可打包成list或dict
    threads = {}
    threads['name1'] = t1
    threads['name2'] = t2
    threads['name3'] = t3
    # 将线程赋予名称
    while 1:
        if str(threads['name1'].isAlive()) == 'False':    # t1挂了
            thread1 = threading.Thread(target=doA,args=(arg1,))   # 传参,target参数预先写好即可
            thread1.start()
            threads['name1'] = thread1         # 更新线程dict,使得重启后的线程可持续监测
        if str(threads['name2'].isAlive()) == 'False':     # t2挂了
            thread2 = threading.Thread(target=doB,args=(arg1,))
            thread2.start()
            threads['name2'] = thread2
        if str(threads['name3'].isAlive()) == 'False':    # t3挂了
            thread3 = threading.Thread(target=doC,args=(arg2,))
            thread3.start()
            threads['name3'] = thread3
        time.sleep(60)    # 自定义监测间隔

实际上我解决的只是down后的线程重启后的持续监测(即加入dict持续监测),对于以执行组件身份来运行的线程,这种方法很不错。至少某一天,api那边出现拥堵后,线程出现了一边挂掉一边重启的样子,警告发了一堆。

然后上面我把重启二字用双引号给标注了,因为这不算真正意义上的重启,当使用isAlive()探测出False时,线程就已经down了,状态都消失了,后面开启的不过是新的线程,然后赋予相同的名字,以持续监测。

发表评论

电子邮件地址不会被公开。 必填项已用*标注