Python线程进程实战：五年经验的踩坑与总结

在我五年的测试开发生涯中，多线程和多进程编程一直是提升程序性能的重要手段。从最初的懵懂踩坑，到现在能够熟练运用各种并发技术，这个过程中积累了大量的实战经验。

今天，我想和大家分享一些在实际项目中遇到的典型问题和解决方案，希望能帮助大家避免一些常见的陷阱，更好地利用Python的并发能力。

Case1： 多进程中，如果遇到闭包如装饰器等函数，进程会报Can‘t pickle local object

参考文档：

https://blog.csdn.net/qq_43331089/article/details/126227288

解决办法：这个问题的根本原因就是，你使用了闭包或者装饰器等功能，多进程在进行序列化时不能将其序列化。一般的解决办法就是避免使用闭包等功能

还提供了一个办法是使用下面的第三方库，但是这个方法我没有使用过

from pathos.multiprocessing import ProcessingPool as Pool

Case2：各种进程线程调用方式

总结了工作中,各种进程线程的写法

线程池

线程池的demo，可以看下他们对参数的调用部分

from concurrent.futures import ThreadPoolExecutor
from common.log_moudle import logger

if __name__ == '__main__':
    with ThreadPoolExecutor(max_workers=10) as executor:
        for i in range(20):
            # executor.submit(case_num,num = i,cas = "1")
            executor.submit(case_num, i,"1")

对进程池的封装

下面的代码分为注释的部分和没有注释的部分，用with的方法更简便，注释掉的部分还会做一个操作就是等待所有任务结束后，才能继续执行后面的代码

参考文档：https://www.cnblogs.com/hahaa/archive/2022/08/03/16547372.html

 ## 封装成通用的函数
 def thread_pool_fun(self,max_workers,func,*args,**kwargs):
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            for i in self.ips:
                executor.submit(func, ip=i, *args, **kwargs)
        
        # executor = ThreadPoolExecutor(max_workers=max_workers)
        # all_task = [ executor.submit(func, ip=i, *args, **kwargs) for i in self.ips ]
        # for future in as_completed(all_task):
        #     data = future.result()
        #     logger.info(f"{data}")
        
        
 ## 调用
     def push_monkey_all(self):
        '''
        把本地的monkey脚本推送到POD中
        '''
        self.thread_pool_fun(1,push_monkey,path = os.path.join(self.monkeyfilepath, self.monkeyfile))

多线程

https://blog.csdn.net/weixin_38848437/article/details/123736285

    def thread_list(self,func,*args,):
        threads = []
        for pod in self.ips:
            logger.info(pod)
            t = threading.Thread(target=func, args=(pod,*args))
            threads.append(t)
        for t in threads:
            time.sleep(0.05)  ## 设置睡眠的时间，防止Max retries exceeded with url
            # t.setDaemon(True)
            t.start()
        for t in threads:
            t.join()

普通的多线程使用方式

    def push_monkey_all(self):
        '''
        把本地的monkey脚本推送到POD中
        '''

        for ip in self.ips:
            t = threading.Thread(target=push_monkey,args=(ip, os.path.join(self.monkeyfilepath, self.monkeyfile),))
            t.start()
            time.sleep(0.1)

多线程还需要注意的地方：

传递参数的方法：
1.使用args 传递参数 threading.Thread(target=sing, args=(10, 100, 100))
2.使用kwargs传递参数 threading.Thread(target=sing, kwargs={“a”: 10, “b”:100, “c”: 100})
3.同时使用 args 和 kwargs 传递参数 threading.Thread(target=sing, args=(10, ), kwargs={“b”: 100,“c”: 100})

多进程

最普通的方式，一般的方式和多线程是一样的

p = Process(target=t.sence_one,args=())
p.start()
p.join()

进程池

    @staticmethod
    def call_back(res):
        logger.info(f' succes {res}')
        # logger.info(f' hello world {res}')

    @staticmethod
    def err_call_back(err):
        logger.debug(f'出错了~error: {str(err)}')
        # logger.info(f'出错了~error: {str(err)}')

def monkey_game_all(self):
        '''
        根据monkey启动所有游戏的
        '''
        pool = Pool(processes=170)
        for ip in self.ips:
                        pool.apply_async(monkey_app_process, args=(ip,), error_callback=self.err_call_back,callback=self.call_back)
        pool.close()
        pool.join()

case3 进程或者线程不退出

在使用monkey的过程中，会发现，如果使用monkey -f xxx.script 111111 -v 1对于这种不退出的命令，我们可以使用 & 符号

例如下面这个代码片段，善于使用 & 会让我省略很多其他的工作，比如我们是不是还要判断进程有没有自动退出，没有退出还要杀掉进程。这块是非常麻烦的

def monkey_run(ip):
    # adb_connect(ip)
    # time.sleep(0.01)
    if ip and len(ip):
        cmd = " /usr/local/bin/adb -s " + ip
    else:
        cmd = " /usr/local/bin/adb "
    # cmd += " shell screencap -p /sdcard/{}.png".format(name)
    cmd += " shell monkey -f  /sdcard/{}  99999 -v 1 &".format(monkeyscript)
    # print(cmd)
    logger.info(cmd)
    f = Popen(cmd, stdout=PIPE, stderr=STDOUT, shell=True)
    # time.sleep(0.1)
    # f.kill()# ignore_security_alert
    resp = f.stdout.readlines()
    # logger.info(f"{cmd}   resp:  {resp}")
    # print(resp)
    return resp

Case4 timer的使用范例

这块的范例是其他人写好

#!/usr/bin/python

import sys
import re
import subprocess
from multiprocessing import Pool
from threading import Timer
# from python_learn.utils.read_file_line import read_line

def read_line(filename):
    res = []
    fp = open(filename, "r")
    for line in fp.readlines():
        res.append(line.strip())
    return res

def timeout_callback(p):
    # print 'pid %d exec timeout, to kill' % p.pid
    try:
        p.kill()
    except Exception as error:
        pass
        # print error


def exec_cmd(remote_host, cmd, timeout=180):

    cmd = "adb -s {0}:5555 shell '{1}'".format(remote_host, cmd)
    # print "[" + remote_host + "] " + cmd

    p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
		# 这块代码的含义就是，过了多少时间后，判断这个进行有没有退出，如果没有，就杀掉进程，如果杀掉进程的过程报错，那么raise异常。
    my_timer = Timer(timeout, timeout_callback, [p])
    my_timer.start()
    try:
        stdout, stderr = p.communicate()
        exit_code = p.returncode
        # print "remote:%s code:%d stdout:%s" % (remote_host, exit_code, stdout)
        if exit_code != 0:
            pass
            # print "stdout:%s stderr:%s" % (stdout, stderr)
    finally:
        my_timer.cancel()


# python batch_cmd.py 'uname -a' | grep '15:30:10' | awk '{print $1}'
# python batch_cmd.py 'setprop ctl.restart devicedaemon'
if __name__ == '__main__':
    if len(sys.argv) < 2:
        print("Usage: " + __file__ + " <cmd>")
        exit(1)

    devices = read_line("devices")

    pool = Pool(128)
    for ip in devices:
        pool.apply_async(exec_cmd, args=(ip, sys.argv[1]))
    pool.close()
    pool.join()
    print("done")

case5： 在主函数中，执行一个线程，当主线程结束时，停止线程

先说这个案例的一个真实需求：

当时测试网络的一个需求时，想要在执行主要的逻辑时，不断的给服务制造一定的压力，这就要写一个后台，让主函数执行的时候，有一个不断循环执行的子任务一直执行，当主函数停止，子任务随着暂停

再说上面脚本的一个弊端：

使用的 event，子任务收到停止的信号后，要执行完子任务的所有步骤，整个主函数才会停止

Event()的使用如下：python threading.Event() 用法

代码如下

import threading


stop_event = threading.Event()

def background_task(func: callable, **kwargs):
    """后台线程执行的操作"""
    print(f"{stop_event.is_set()}")

    while not stop_event.is_set():
        print(f"{stop_event.is_set()}")
        print("Background task is running...")
        func(**kwargs)
        time.sleep(1)


def background_dump(port):
    # 测试 sp 时，制造压力的一个线程，不断 dump 数据的操作
    print("dumping")
    # 执行 grpcurl 命令
    resp = cmd_comand.terminal_shell(
        f"ssh root@{bpa_ip} 'bash /root/grpcurl_demo.sh dump {port}'"
    )
    # logger.info(resp)
    # 将结果写入文件
    with open("dump.txt", "a") as f:
        f.write(resp)
    logger.info(f"{port}")

def mock_test():
    global stop_event
    # 启动后台线程
    thread = threading.Thread(target=background_task,kwargs={"func":background_dump, "port":50051})
    thread.start()

    # 在这里执行主函数的其他操作
    print("Main function is executing...")
    time.sleep(5)  # 模拟主函数的工作时间
    print("Main function is done.")

    # 完成后设定停止事件
    stop_event.set()

    # # 等待线程结束
    # thread.join()
    print("Background thread has been stopped.")

调用方式

Event其实就是一个简化版的 Condition。Event没有锁，无法使线程进入同步阻塞状态。

isSet(): 当内置标志为True时返回True。

set(): 将标志设为True，并通知所有处于等待阻塞状态的线程恢复运行状态。

clear(): 将标志设为False。

wait([timeout]): 如果标志为True将立即返回，否则阻塞线程至等待阻塞状态，等待其他线程调用set()。

if __name__ == '__main__':
    mock_test()