目录

1. 概述:

2. 进程概念：

4. 进程池和线程池 

5. python进程代码实现

 5.1 基础样例

5.2 杀掉某个进程中的所有子进程方法

6. python线程代码实现

        6.1 普通写法

        6.2 重构方法：

7. 进程间变量共享

        7.1 使用共享内存（Value/Array）

        7.2使用Manager管理共享对象

         7.3 同步与锁机制

        7.4 使用Queue和Pipe进行进程间通信

          7.5 如果需要共享多个变量，

8. 线程间变量共享

        8.1 全局变量 + 锁

        8.2队列 (Queue)

       8.3 线程局部变量 (threading.local())

9. redis的应用--多进程和多线程变量共享中

python中的协程使用

---

1. 概述:

本文了解线程和进程概念，使用方法，变量共享

线程进程概念并非仅为python独有的，在计算机的概念中都比较实用重要。

很多场景都需要用到并发，如测试框架，用例执行时同步需抓串口日志，pcap包等。或者一种简单的任务需要多次运行如1千，1万次，且具备同时运行的条件，多进程线程就可以派上用场，大幅度节省时间。或者在一些性能稳定性相关测试，高并发的场景并不罕见，都需要用到。

2. 相关概念：

        2.1. 进程概念： 
        程序执行的实例，包含虚拟地址空间、执行代码和系统资源。（如手机的一个独立app运行，就至少启了一个进程） 
        2.2 . 线程概念： 
        是操作系统能够进行运算调度的最小单位 
        进程内执行单元，共享进程资源，至少包含程序计数器、寄存器组和线程栈。 
        可以理解为，如果有八个线程，那么八个线程都在同一个进程内，以非常小的时间依次替换运行，一个线程在运行时，其他线程进程等待阶段，直至资源释放后再执行。 
        2.3 协程概念 
        协程是一种用户态的轻量级线程，它允许函数在执行过程中暂停（yield）并在之后恢复（resume），从而在不阻塞线程的情况下实现并发任务

3.进程线程协程对比

      

4. 进程池和线程池 

以下用concurrent.futures 模块提供的线程池来管理和执行多个线程任务。线程池提供了一种方便的方式来处理并发任务，可以有效地利用系统资源，提高程序的性能。并将每个进程的结果，汇到列表中。

# 示例: 使用线程池执行了5个任务，并通过as_completed()函数等待任务完成并获取结果。最后，打印出所有任务的结果。
import concurrent.futures 
def task(name):
    print(f"线程 {name}")
    return name 
if __name__ == '__main__':
    with concurrent.futures.ThreadPoolExecutor() as executor:  # 创建线程池，默认使用CPU核心数作为线程数
        futures = [executor.submit(task, i) for i in range(5)]      # 提交5个任务给线程池  executor.submit(task, 1) 提交任务给线程池：future = executor.submit(function, *args, **kwargs)  调用submit()方法提交一个任务给线程池。function是要执行的函数，*args和**kwargs是函数的参数。
        results =  []
        for future in concurrent.futures.as_completed(futures):
            results.append(future.result())  # 获取任务的结果  通过result()方法获取任务的执行结果。如果任务还没有完成，则该方法会阻塞直到任务完成并返回结果。
        print(results) # 打印结果

5. python进程代码实现

 5.1 基础样例

       multiprocessing  多进程使用的类

        Process  是上述类的一个方法

        ptt.join() 等待进程结束，

        p4.terminate() # 第二个是主动关闭子进程（如一个独立抓串口日志的进程,如果一直循环的话，就需要手动关掉进程）

from multiprocessing import Process
import time
class G:
    def a(self,ff="aaaaa",max_i=3):
        for i in range(max_i):
            print(ff)
            time.sleep(0.3) 
    def b(self,ff="aaaaa",max_i=3):  # 结构虽和函数a相同，写法仅为下述其多进程理解
        for i in range(max_i):
            print(ff)
            time.sleep(0.3)  
    def run(self):
        p1 = Process(target = self.a, args = ("进程1",2,))  # 打印两次 进程1
        p2 = Process(target = self.a, args = ("进程2",3,)) # 打印三次 进程2
        p3 = Process(target = self.b, args = ("进程3",5,)) # 打印五次 进程3
        p4 = Process(target = self.b, args = ("进程4",50000,)) # 打印50000次 进程4
        for ppp in [p1, p2, p3, p4]: #启动四个进程
            ppp.start()  
        for ptt in [p1, p2, p3]:  #主进程等待三个进程自动运行结束
            ptt.join()   
        p4.terminate() # 主进程运行到这里会，强制结束 进程p4， 
if __name__ == '__main__': 
   G().run()

5.2 杀掉某个进程中的所有子进程方法

        如抓串口日志的是一个进程，但这个进程内抓了8个串口，就需要开八个子进程

import psutil, signal
class 。。。。
     # 工具函数 """终止指定PID的进程及其所有子进程"""
    def kill_process_tree(self, pid, sig = signal.SIGTERM):   
        try:
            parent = psutil.Process(pid)  # 递归获取所有子进程
            children = parent.children(recursive = True)  # 先终止子进程
            for child in children: #  print(f"Terminating child PID {child.pid}")
                child.kill()  # 发送SIGKILL强制终止 
            print(f"Terminating parent PID {pid}")   # 最后终止父进程
            parent.kill()
        except psutil.NoSuchProcess:
            print(f"Process {pid} does not exist")
        except psutil.AccessDenied:
            print(f"Permission denied for process {pid}")    

6. python线程代码实现

        6.1 普通写法

import threading ,time 
def worker(num):
    print(f"线程 {num} 开始")
    time.sleep(1)
    print(f"线程 {num} 结束") 
t1 = threading.Thread(target=worker, args=(1,))# 创建线程
t2 = threading.Thread(target=worker, args=(2,))# 创建线程  
t1.start()# 启动线程
t2.start()  
t1.join()# 等待所有线程完成
t2.join()
print("所有线程执行完毕")

        6.2 重构方法：

        threading 线程使用的类，下述方法使用的是重构，只调用run()函数，多个相同函数时仅最后一个run()生效

import threading,time 
class MyThread(threading.Thread):
    def __init__(self, name):
        super().__init__()
        self.name = name 
    def run(self):
        for i in range(5):
            print(f"线程 {self.name} 正在运行")
            time.sleep(0.1)  
    def run3(self):
        print(999999999999999)
t = MyThread("MyThread1")
t.start()
t.join()

7. 进程间变量共享

        7.1 使用共享内存（Value/Array）

Value和 Array通过共享内存直接实现数据共享，效率较高

import multiprocessing 
def worker(num, arr): 
    num.value += 1# 修改共享变量的值
    arr[0] += 10 
if __name__ == '__main__': 
    shared_number = multiprocessing.Value('i', 0) # 创建共享整数（'i' 表示整数类型，初始值为0） 
    shared_array = multiprocessing.Array('i', [1, 2, 3, 4, 5])  # 创建共享数组（'i' 表示数组元素为整数，初始序列为 [1, 2, 3, 4, 5]） 
    p = multiprocessing.Process(target=worker, args=(shared_number, shared_array))
    p.start()
    p.join() 
    print(f"shared_number.value: {shared_number.value}")  #运行结果：shared_number.value: 1
    print(f"shared_array[:]: {shared_array[:]}")  #运行结果：#shared_array[:]: [11, 2, 3, 4, 5]

        7.2使用Manager管理共享对象

Manager可以创建共享的列表、字典等更复杂的Python对象，使用起来更直观，但速度通常比 Value和 Array慢

import multiprocessing 
def worker(shared_dict, shared_list, index):
    shared_dict[index] = index * index  # 向字典添加键值对
    shared_list[index] = index * 2      # 修改列表元素 
if __name__ == '__main__':
    with multiprocessing.Manager() as manager: 
        my_dict = manager.dict()# 创建共享字典和列表
        my_list = manager.list([0] * 5)  # 初始化一个长度为5的列表 
        processes = []
        for i in range(5):
            p = multiprocessing.Process(target=worker, args=(my_dict, my_list, i))
            processes.append(p)
            p.start() 
        for p in processes:
            p.join() 
        print(f"共享字典: {my_dict}") # 运行结果 ：共享字典: {0: 0, 1: 1, 2: 4, 4: 16, 3: 9}
        print(f"共享列表: {my_list}") # 运行结果 ： 共享列表: [0, 2, 4, 6, 8]

         7.3 同步与锁机制

当多个进程可能同时修改同一个共享变量时，可能产生竞争条件（Race Condition）。使用锁（Lock）可以确保同一时刻只有一个进程能修改数据

import multiprocessing 
def safe_increment(counter, lock):
    for _ in range(1000):
        with lock:  # 使用 with 语句自动获取和释放锁
            counter.value += 1 
if __name__ == '__main__':
    counter = multiprocessing.Value('i', 0)
    lock = multiprocessing.Lock() 
    processes = [multiprocessing.Process(target=safe_increment, args=(counter, lock)) for _ in range(4)] 
    for p in processes:
        p.start()
    for p in processes:
        p.join() 
    print(f"Final counter value: {counter.value}")  # 应该是 4000

        7.4 使用Queue和Pipe进行进程间通信

除了共享内存，还可以通过消息传递（进程间通信，IPC）来共享数据。Queue（多生产者、多消费者）和 Pipe（点对点）是常用方式

import multiprocessing 
def producer(queue, data):
    for item in data:
        queue.put(item)  # 将数据放入队列 
def consumer(queue, name):
    while True:
        item = queue.get()  # 从队列取出数据
        if item is None:  # 使用特殊信号表示结束
            break
        print(f"{name} processed: {item}") 
if __name__ == '__main__':
    my_queue = multiprocessing.Queue() 
    prod = multiprocessing.Process(target=producer, args=(my_queue, [1, 2, 3, 4, 5]))
    cons = multiprocessing.Process(target=consumer, args=(my_queue, "Consumer")) 
    prod.start()
    cons.start() 
    prod.join()
    my_queue.put(None)  # 发送结束信号给消费者
    cons.join() 

          7.5 如果需要共享多个变量，

        可以将以上代码 my_queue = multiprocessing.Queue()

        下方新增一个Queue()即可如,变量正常调用:   

        my_queue = multiprocessing.Queue() 

        my_queue_2 = multiprocessing.Queue()

8. 线程间变量共享

        8.1 全局变量 + 锁

这是最基础的方法，关键在于使用锁来保证操作的原子性。 

import threading
shared_counter = 0  # 全局变量
lock = threading.Lock()  # 创建锁对象
def increment():
    global shared_counter
    for _ in range(1000):
        with lock:  # 使用with语句自动获取和释放锁
            shared_counter += 1
threads = []
for i in range(5):
    t = threading.Thread(target=increment)
    threads.append(t)
    t.start()
for t in threads:
    t.join() 
print(f"Final counter: {shared_counter}")  # 结果是5000

​何时选择​：适用于共享数据结构简单、修改操作快速的场景。

        8.2队列 (Queue)

queue.Queue是一个线程安全的先进先出（FIFO）数据结构，是实现生产者-消费者模型的理想选择。 

import threading
import queue
import time
task_queue = queue.Queue()  # 创建线程安全队列
def producer(): #  """生产者线程：放入任务"""
    for i in range(5):
        time.sleep(0.1)  # 模拟生产耗时
        task_queue.put(f"Task-{i}")
        print(f"Produced: Task-{i}")
    task_queue.put(None)  # 发送结束信号
def consumer(): #   """消费者线程：取出任务并处理"""
    while True:
        item = task_queue.get()
        if item is None:  # 收到结束信号
            break
        print(f"Consumed: {item}")
        task_queue.task_done()  # 告知任务已完成
prod_thread = threading.Thread(target=producer)
cons_thread = threading.Thread(target=consumer)
prod_thread.start()
cons_thread.start()
prod_thread.join()
cons_thread.join()

​何时选择​：当线程间有明显分工，一方产生数据，另一方处理数据时，应优先使用队列。它能有效解耦线程，平衡处理速度

       8.3 线程局部变量 (threading.local())

当需要每个线程拥有自己独立的变量，避免相互干扰时，可以使用 threading.local()。

import threading 
local_data = threading.local()# 创建线程本地存储对象 
def worker(thread_id): 
    local_data.value = f"Value-from-{thread_id}"    # 每个线程设置自己独有的值 
    print(f"Thread {thread_id}: {local_data.value}")  # 在同一线程内，可以安全地访问和修改 local_data 的属性
threads = []
for i in range(3):
    t = threading.Thread(target=worker, args=(i,))
    threads.append(t)
    t.start() 
for t in threads:
    t.join()
#运行结果
#Thread 0: Value-from-0
#Thread 1: Value-from-1
#Thread 2: Value-from-2

9. redis的应用--多进程和多线程变量共享中

        一遍多进程和多线程还可以采用写入json，csv，text等文件方式，但是效率比较慢，又容易出现资源竞争的问题，不够准确。而采用redis绝对是很好的方法，特别是共享数据比较大时。一般成体系的测试框架也可以建议使用。

以下是一份示例，更多高级方法后面再补充。

import redis，json， pickle
from datetime import timedelta 
# 1. 基本连接
r = redis.Redis( host='localhost',
    port=6379,  db=0, password=None,  # 如果有密码请设置
    decode_responses=True,  # 自动解码为字符串
    socket_connect_timeout=5 ) # 连接超时时间  
try:
    r.ping()# 测试连接
    print("✅ Redis连接成功")
except redis.ConnectionError as e:
    print(f"❌ Redis连接失败: {e}")
    exit(1) 
# 2. 使用连接池（生产环境推荐）
pool = redis.ConnectionPool(   host='localhost',  port=6379,
    db=0,  max_connections=20,  decode_responses=True )
r_pool = redis.Redis(connection_pool=pool)

class RedisCacheSystem:
    def __init__(self, redis_client, namespace="app_cache"):
        self.redis = redis_client
        self.namespace = namespace 
    def make_key(self, key):
        return f"{self.namespace}:{key}" 
    def set(self, key, value, expire_seconds=None):
        cache_key = self.make_key(key)
        serialized_value = pickle.dumps(value) 
        if expire_seconds:
            self.redis.setex(cache_key, expire_seconds, serialized_value)
        else:
            self.redis.set(cache_key, serialized_value) 
    def get(self, key):
        cache_key = self.make_key(key)
        serialized_value = self.redis.get(cache_key) 
        if serialized_value:
            return pickle.loads(serialized_value)
        return None 
    def delete(self, key):
        cache_key = self.make_key(key)
        self.redis.delete(cache_key) 
    def clear_namespace(self):
        """清除命名空间下的所有键"""
        pattern = f"{self.namespace}:*"
        keys = self.redis.keys(pattern)
        if keys:
            self.redis.delete(*keys) 
# 使用示例
cache_system = RedisCacheSystem(r)
cache_system.set("user_profile", {"name": "John", "preferences": {"theme": "dark"}}, 3600)
profile = cache_system.get("user_profile")
print("缓存数据:", profile)

10.python中的协程使用

从实际应用中了解

图中AsyncSniffer 是 一个非常强大的网络库scapy的一个方法，作用是异步监控某个网口的报文。

协程的优势在于：

         高并发I/O操作：在处理大量网络连接或文件读写时，协程是理想选择。当一个协程等待I/O响应而暂停时，线程可以立即切换到其他就绪的协程继续工作，用少量线程支撑海量并发任务，极大提升资源利用率

        如爬虫等，高度使用http请求的也非常适合使用协程，无脑用进程线程会影响其效率和资源占用。

import asyncio 
async def say_after(delay, what):
    await asyncio.sleep(delay)
    print(what) 
async def main():
    # 创建任务，它们会被事件循环并发调度
    task1 = asyncio.create_task(say_after(1, "Hello"))
    task2 = asyncio.create_task(say_after(2, "World")) 
    # 等待两个任务都完成
    await task1
    await task2  # 总耗时约2秒，而非串行的3秒 
asyncio.run(main())
#运行结果
#Hello
#World