第21篇多线程架构与线程安全 — 并发请求的正确处理姿势一、多线程模型概述1.1 为什么需要多线程在深入讨论GrainServer的多线程架构之前我们首先要回答一个根本问题为什么需要多线程单线程不行吗对于后端服务来说单线程模式有一个致命的缺陷串行处理请求。当服务器正在处理一个请求时其他所有请求都必须排队等待。如果某个请求耗时很长比如晶粒分析的模型推理可能需要数秒后续的请求就会被阻塞用户体验极差。多线程的核心价值在于通过并发执行提升系统的吞吐量和响应性。具体到GrainServer项目多线程的必要性体现在以下几个方面双服务器并行运行GrainServer有两个Socket服务器端口5000和5001分别处理初始分析和重计算任务。如果是单线程两个服务器无法同时工作必须等一个处理完才能轮到另一个。用多线程可以让两个服务器独立运行互不干扰。多客户端同时连接每个Socket服务器可能同时接受多个客户端连接。如果是单线程一个客户端的长任务会阻塞其他所有客户端。每连接一线程的模式可以让多个客户端并行处理。I/O等待利用Socket通信、文件读写、图像加载等操作都包含大量的I/O等待时间。在单线程模式下CPU在I/O等待时只能闲置。多线程可以让CPU在一个线程等待I/O时切换到另一个线程执行计算提高CPU利用率。响应性保证即使有一个任务因为某种原因卡住了比如模型推理超时也不会影响其他任务的执行系统整体仍然可用。当然多线程不是银弹。它带来了并发能力的同时也引入了线程安全、死锁、资源竞争等一系列复杂问题。这正是本文要深入探讨的主题。1.2 双服务器架构GrainServer采用了双Socket服务器的架构设计两个服务器分别监听不同的端口处理不同类型的任务服务器端口客户端标识功能说明Server15000handle接收原始图像执行模型推理和初始粒径计算Server25001cal接收人工修正后的图像重新进行粒径计算跳过模型推理这样的架构设计有几个明显的优势职责分离初始分析和重计算是两种不同的业务场景分开处理逻辑更清晰代码更容易维护。资源隔离重计算任务不需要运行深度学习模型资源占用较少。两种任务用不同的服务器处理可以避免重计算任务被耗时的模型推理任务阻塞。故障隔离如果某个服务器因为bug崩溃了另一个服务器还能继续工作不会导致整个服务完全不可用。扩展性未来如果需要增加新的任务类型如批量处理、实时预览等可以继续增加新的服务器端口不会影响现有功能。每个服务器都运行在独立的线程中两者并行不悖。这是GrainServer多线程架构的第一层。1.3 每连接一线程模型在每个Socket服务器内部GrainServer采用了每连接一线程Thread-per-connection的模型。每当有新的客户端连接进来服务器就会创建一个新的线程来处理这个客户端的所有请求。这种模型的特点是实现简单每个线程只需要处理一个连接逻辑是线性的不需要复杂的状态机。隔离性好一个连接的崩溃或异常不会影响其他连接。响应及时每个连接都有独立的线程不会因为其他连接的繁忙而得不到响应。当然每连接一线程也有其局限性线程创建开销每次连接都创建新线程有一定的创建和销毁开销。对于短连接较多的场景开销会比较明显。线程数量不可控如果同时有大量客户端连接线程数量会急剧膨胀可能耗尽系统资源。上下文切换开销线程数量太多时CPU在不同线程之间切换的开销会显著增加反而降低整体性能。对于GrainServer这种应用场景来说每连接一线程是完全合适的。因为客户端数量不多通常是一个前端程序对应一个连接每个连接的生命周期较长不是频繁建立和断开的短连接任务以计算为主线程数量不会成为瓶颈这是GrainServer多线程架构的第二层。1.4 两层多线程的叠加将双服务器和每连接一线程结合起来GrainServer的整体多线程结构如下主线程 ├── Server1线程 (端口5000) │ ├── 客户端连接线程1 │ ├── 客户端连接线程2 │ └── ... └── Server2线程 (端口5001) ├── 客户端连接线程1 ├── 客户端连接线程2 └── ...主线程负责启动两个服务器线程然后等待。每个服务器线程负责监听端口、接受连接并为每个连接创建新的处理线程。这种两层多线程的架构既保证了两个服务器的独立运行又保证了每个客户端连接的并发处理是一种简洁而有效的设计。二、ThreadLocalStorage线程局部存储原理2.1 为什么不用全局变量在多线程环境中一个最容易犯的错误就是使用全局变量来存储每个线程的私有数据。我们来看TaskHandle.py中被注释掉的代码# self.data_dict {# metal: [],# grain_count: [],# ...# }# self.path None# self.metal None# self.filename None这段代码来自f:\grainserver\SocketServ\TaskHandle.py:62-86。可以看到在代码的早期版本中data_dict、path、metal、filename这些变量是作为SocketServTaskHandle类的实例属性存在的。但这在多线程环境下会有严重的问题。让我们设想这样一个场景线程A正在处理任务1设置了self.metal Tiself.path ...开始运行模型推理模型推理需要几秒钟时间就在这几秒钟内线程B也开始处理任务2设置了self.metal Alself.path ...线程A的模型推理结束继续执行后续的后处理但此时self.metal已经被线程B改成了Al结果任务1的金属类型被错误地记录为Al数据混乱这就是典型的线程安全问题——多个线程同时读写共享的数据导致数据不一致。问题的根源在于SocketServTaskHandle是一个单例对象整个程序只有一个实例但它的属性被多个线程共享使用。每个线程都在修改同一个对象的属性互相覆盖最终导致数据错乱。2.2 什么是线程局部存储既然全局变量包括实例属性不能用那每个线程需要的私有数据应该放在哪里答案就是线程局部存储Thread Local Storage, TLS。线程局部存储的核心思想是为每个线程提供一份独立的数据副本线程之间互不干扰。打个比方全局变量就像一个公共的白板所有人都在上面写字你刚写的内容可能转眼就被别人擦掉了线程局部存储就像每个人都有自己的私人笔记本你在自己的本子上写什么别人都看不到也改不了从技术实现上来说线程局部存储通常是一个字典以线程ID为键每个线程对应自己的数据。当线程访问TLS变量时系统会自动根据当前线程的ID找到对应的数据副本。Python标准库提供了threading.local()类来实现线程局部存储使用非常简单importthreading local_datathreading.local()defworker(name):local_data.namename# ... 做一些事情 ...print(local_data.name)# 每个线程看到的都是自己设置的值每个线程对local_data的属性赋值都只在自己的线程内可见不会影响其他线程。2.3 GrainServer中的ThreadLocalStorage在GrainServer项目中线程局部存储被封装为一个ThreadLocalStorage类继承自threading.localclassThreadLocalStorage(threading.local):def__init__(self):super().__init__()self.data_dict{metal:[],grain_count:[],grain_size:[],grain_0_30:[],grain_30_60:[],grain_60_90:[],grain_90_120:[],grain_120_150:[],grain_150_180:[],grain_180_210:[],grain_210_240:[],grain_240_270:[],grain_270_300:[],grain_300:[],grain_min:[],grain_max:[],grain_std:[]}self.pathNoneself.metalNoneself.filenameNone这段代码来自f:\grainserver\SocketServ\TaskHandle.py:20-44。让我们仔细分析这个类的设计继承自threading.local通过继承threading.localThreadLocalStorage自动获得了线程局部存储的能力。每个线程访问这个对象的属性时得到的都是自己的副本。初始化方法中定义属性在__init__方法中初始化了所有需要的属性。注意这里的初始化是在每个线程第一次访问这个对象时执行的确保每个线程都有自己的初始值。data_dict字典这是一个嵌套的数据结构包含了所有粒径统计结果字段。每个字段的值都是一个列表虽然实际使用时列表中只有一个元素。三个简单属性path、metal、filename分别存储当前处理的图像路径、金属类型和文件名。在SocketServTaskHandle类中TLS对象被创建为实例属性self.thread_localThreadLocalStorage()这段代码来自f:\grainserver\SocketServ\TaskHandle.py:81。2.4 ThreadLocalStorage的使用方式在任务处理过程中所有的线程私有数据都通过self.thread_local来访问而不是直接使用self的属性。例如在handle_client方法中ifmetalinself.metals:self.thread_local.pathpath self.thread_local.filenamefilename self.thread_local.metalmetal self.handle_metal_model_request(client_socket)这段代码来自f:\grainserver\SocketServ\TaskHandle.py:119-123。设置当前任务的路径、文件名和金属类型时都是赋值给self.thread_local的属性而不是self的属性。在后续的处理中访问这些数据也是通过self.thread_localself.logger.info(fHandle{self.thread_local.metal}picture from{client_socket})这段代码来自f:\grainserver\SocketServ\TaskHandle.py:217。这样无论有多少个线程同时在处理任务每个线程看到的self.thread_local.metal都是自己设置的值不会被其他线程覆盖。2.5 threading.local()的实现原理你可能会好奇threading.local()是怎么做到同一个对象不同线程看到不同属性的它的底层原理是什么虽然Python的threading.local是用C实现的性能更好但我们可以用纯Python代码来模拟它的工作原理帮助理解importthreadingclassSimpleLocal:def__init__(self):self._data{}def__getattr__(self,name):thread_idthreading.get_ident()ifthread_idnotinself._data:raiseAttributeError(fThread{thread_id}has no attribute {name})returnself._data[thread_id][name]def__setattr__(self,name,value):ifname_data:super().__setattr__(name,value)returnthread_idthreading.get_ident()ifthread_idnotinself._data:self._data[thread_id]{}self._data[thread_id][name]value简化版的原理就是内部维护一个字典_data以线程ID为键每个线程对应一个子字典存储该线程的所有属性当读取属性时根据当前线程ID找到对应的子字典再取值当设置属性时同样根据线程ID找到或创建子字典再存值真实的threading.local比这个复杂得多比如要处理线程退出后的清理、弱引用、子类化等问题但核心思想是一样的。三、互斥锁与临界区3.1 什么时候需要锁线程局部存储解决了每个线程私有数据的问题但并不是所有数据都能放在TLS里。有些数据是需要在线程之间共享的这时候就需要用**锁Lock**来保证线程安全。等等你可能会问GrainServer中的data_dict不是已经放在TLS里了吗每个线程有自己的副本那为什么还需要锁让我们来看write_data_dict方法defwrite_data_dict(self):withself.lock:# ... 操作 self.thread_local.data_dict ...这段代码来自f:\grainserver\SocketServ\TaskHandle.py:323-324。确实data_dict是线程局部的每个线程操作的都是自己的副本从这个角度说不需要锁。但是write_data_dict方法中除了操作data_dict还做了其他事情调用self.ImgUtils.plot_length_distribution()绘图调用self.FileHandle.write_dict_to_txt()写文件这些操作涉及到共享资源的访问Matplotlib的绘图状态在某些情况下是全局共享的文件系统是共享资源虽然写的是不同的文件但文件系统本身有底层锁日志系统也是共享的更重要的是with self.lock确保了write_data_dict方法中的所有操作是原子的——一个线程要么完整地执行完整个方法要么完全不执行不会执行到一半被另一个线程打断。这引出了一个重要的概念临界区Critical Section。临界区是指一段访问共享资源的代码这段代码在任意时刻只能被一个线程执行。锁的作用就是保护临界区确保并发安全。3.2 threading.Lock()的工作原理Python的threading.Lock是最基本的互斥锁Mutex它有两种状态已锁定locked和未锁定unlocked。锁提供了两个基本操作acquire()获取锁。如果锁是未锁定的就将其改为锁定并立即返回如果锁已经被锁定就阻塞等待直到锁被释放。release()释放锁。将锁的状态改为未锁定唤醒一个正在等待的线程。在Python中最推荐的锁使用方式是with语句withself.lock:# 临界区代码with语句会自动在进入时获取锁在退出时释放锁即使临界区代码抛出异常锁也会被正确释放。这比手动调用acquire()和release()更安全也更简洁。GrainServer中就是这样使用的self.lockthreading.Lock()这段代码来自f:\grainserver\SocketServ\TaskHandle.py:82。defwrite_data_dict(self):withself.lock:# ... 临界区代码 ...这段代码来自f:\grainserver\SocketServ\TaskHandle.py:323-324。3.3 write_data_dict()临界区分析现在让我们深入分析write_data_dict方法看看这个临界区里到底做了什么defwrite_data_dict(self):withself.lock:try:forkeyinself.thread_local.data_dict.keys():self.thread_local.data_dict[key][]self.logger.info(finit data_dict:{self.thread_local.data_dict})interval_keylist(self.DataProc.interval_counts.values())self.logger.info(fdata_dict:{self.thread_local.data_dict})self.thread_local.data_dict[metal].append(self.thread_local.metal)self.thread_local.data_dict[grain_count].append(self.DataProc.total_segments)self.thread_local.data_dict[grain_size].append(self.DataProc.average_length)self.thread_local.data_dict[grain_0_30].append(interval_key[0])self.thread_local.data_dict[grain_30_60].append(interval_key[1])# ... 更多字段赋值 ...self.ImgUtils.plot_length_distribution(self.DataProc.interval_counts,self.DataProc.plot_output_path)self.logger.info(fdata_dict:{self.thread_local.data_dict})self.FileHandle.write_dict_to_txt(data_dictself.thread_local.data_dict,output_fileself.DataProc.dict_path)self.logger.info(fdistribution plot done)exceptExceptionase:self.logger.error(fError processing image:{e})raise这段代码来自f:\grainserver\SocketServ\TaskHandle.py:323-358。让我们逐段分析第一步清空data_dictforkeyinself.thread_local.data_dict.keys():self.thread_local.data_dict[key][]将data_dict中每个字段的列表清空为填入新数据做准备。第二步获取区间计数interval_keylist(self.DataProc.interval_counts.values())从DataProc对象中获取各粒径区间的计数结果。第三步填充数据self.thread_local.data_dict[metal].append(self.thread_local.metal)self.thread_local.data_dict[grain_count].append(self.DataProc.total_segments)# ... 更多字段 ...将各种统计数据填入data_dict对应的字段中。第四步绘制分布图self.ImgUtils.plot_length_distribution(self.DataProc.interval_counts,self.DataProc.plot_output_path)调用ImgUtils绘制粒径分布折线图。第五步写入结果文件self.FileHandle.write_dict_to_txt(data_dictself.thread_local.data_dict,output_fileself.DataProc.dict_path)将结果字典写入文本文件。整个过程被with self.lock:包裹确保同一时间只有一个线程在执行这些操作。3.4 锁的粒度问题这里有一个值得讨论的设计问题锁的粒度是不是太大了把整个write_data_dict方法都放在锁里面意味着线程A在绘图的时候线程B连更新自己的data_dict都不行必须等线程A在写文件的时候线程B也必须等着这会不会影响并发性能要回答这个问题我们需要权衡粗粒度锁当前方案的优点实现简单不容易出错锁的获取和释放次数少开销小不会出现死锁只有一把锁细粒度锁的优点并发度高不同的操作可以并行执行锁的持有时间短对于GrainServer的应用场景来说粗粒度锁是完全可以接受的因为write_data_dict的执行时间很短相对于模型推理的数秒钟数据整理、绘图、写文件加起来也只有几十毫秒锁的持有时间很短。并发量不高同时处理的任务数量不多锁竞争不激烈。简单可靠最重要对于后端服务来说正确性和稳定性比极致的性能更重要。粗粒度锁虽然牺牲了一点并发性能但大大降低了出错的概率。这是一个典型的过早优化是万恶之源的例子。在并发量不大的情况下用简单的粗粒度锁就足够了没必要为了理论上的性能提升而引入复杂的细粒度锁方案。四、死锁风险与规避4.1 什么是死锁死锁Deadlock是多线程编程中最经典的问题之一。当两个或多个线程互相等待对方释放资源而谁都无法继续执行时就发生了死锁。死锁的发生需要同时满足四个必要条件称为死锁四条件互斥条件资源不能被共享同一时间只能被一个线程占用持有并等待线程已经持有了至少一个资源同时又在等待其他线程持有的资源不可剥夺资源只能被持有者主动释放不能被强行夺走循环等待线程之间形成了循环等待的链条每个线程都在等下一个线程持有的资源这四个条件缺一不可只要打破其中任意一个死锁就不会发生。4.2 GrainServer中的死锁风险分析让我们来审视GrainServer的代码看看有没有潜在的死锁风险。首先GrainServer中只有一把锁self.lock。只有一把锁的情况下会不会发生死锁答案是不会因为多把锁的顺序问题而死锁但可能因为同一把锁的重复获取而死锁。等等同一把锁也会死锁如果一个线程在已经持有锁的情况下再次尝试获取同一把锁会发生什么这取决于锁的类型普通锁Lock会死锁。因为普通锁不允许同一个线程多次获取第二次获取会阻塞而锁又被自己持有永远不会被释放于是自己把自己死锁了。可重入锁RLock不会死锁。可重入锁允许同一个线程多次获取只要最后释放相应的次数即可。GrainServer使用的是threading.Lock()也就是普通锁。那么代码中有没有可能在持有锁的情况下再次调用需要获取锁的方法让我们检查调用链write_data_dict持有锁 → 在锁内部调用了self.logger.info(...)→ Logger类内部有没有锁logging模块本身是线程安全的内部有自己的锁但那是另一把锁self.ImgUtils.plot_length_distribution(...)→ 这个方法里有没有获取self.lock没有self.FileHandle.write_dict_to_txt(...)→ 这个方法里也没有获取self.lock看起来write_data_dict内部的调用都不会再次获取self.lock所以不会发生自己把自己锁死的情况。这是好消息。但我们还是要保持警惕因为代码是不断演化的。如果未来有人在plot_length_distribution或者write_dict_to_txt内部也加上了with self.lock死锁就会立刻发生。4.3 死锁的规避策略虽然当前代码中没有明显的死锁风险但了解死锁的规避策略仍然是必要的。以下是几种常用的死锁预防方法策略一固定加锁顺序如果必须使用多把锁所有线程都按照相同的顺序获取锁。例如永远先获取锁A再获取锁B不要有的线程先A后B有的先B后A。这样就打破了循环等待条件。策略二使用可重入锁如果存在递归调用或者嵌套获取同一把锁的场景使用threading.RLock()代替threading.Lock()。可重入锁允许同一个线程多次获取只要释放次数和获取次数一致即可。策略三超时机制获取锁的时候设置超时时间如果超时还没获取到就放弃并释放已经持有的锁。Lock.acquire(timeout5)可以实现这个功能。这样即使发生了死锁也能在超时后自动解开。策略四避免持有锁时调用外部方法在持有锁的情况下尽量不要调用外部的、不确定会不会也加锁的方法。如果必须调用要清楚地知道调用链上会不会获取其他锁。策略五减少锁的使用最好的策略其实是能不用锁就不用锁。比如用线程局部存储、用消息队列、用原子操作等方式来替代锁。没有锁自然也就没有死锁。对于GrainServer来说目前只有一把锁而且使用方式比较简单死锁风险很低。但随着功能的增加未来可能会引入更多的锁到时候就需要注意这些问题了。五、ThreadLocalStorage类代码精讲5.1 类的定义与继承让我们再回过头来逐行精读ThreadLocalStorage类的代码classThreadLocalStorage(threading.local):这段代码来自f:\grainserver\SocketServ\TaskHandle.py:20。ThreadLocalStorage继承自threading.local。这是整个类的核心——通过继承我们获得了线程局部存储的能力。Python的继承机制在这里发挥了作用ThreadLocalStorage的实例会拥有threading.local的所有特性同时我们可以在子类中添加自己的属性和方法。5.2 初始化方法def__init__(self):super().__init__()self.data_dict{metal:[],grain_count:[],grain_size:[],# ... 更多字段 ...grain_min:[],grain_max:[],grain_std:[]}self.pathNoneself.metalNoneself.filenameNone这段代码来自f:\grainserver\SocketServ\TaskHandle.py:21-44。初始化方法中有几个关键点调用父类初始化super().__init__()调用父类threading.local的初始化方法。这是子类化时的标准做法确保父类的逻辑被正确执行。每个线程独立初始化你可能会以为__init__只在创建对象的时候执行一次但对于threading.local的子类来说__init__会在每个线程第一次访问这个对象时都执行一次。这样每个线程都能获得自己的初始值。data_dict的结构data_dict是一个字典每个键对应一个统计字段每个值都是一个列表。注意虽然值是列表但实际使用时列表中只有一个元素。为什么用列表而不是直接存值这可能是为了与FileHandle.write_dict_to_txt方法的格式要求保持一致——那个方法期望每个值都是列表。三个简单属性path、metal、filename是三个简单的标量属性存储当前任务的基本信息。5.3 在SocketServTaskHandle中的使用在SocketServTaskHandle类中TLS对象被创建为实例属性self.thread_localThreadLocalStorage()这段代码来自f:\grainserver\SocketServ\TaskHandle.py:81。注意虽然self.thread_local是SocketServTaskHandle的实例属性被所有线程共享但因为ThreadLocalStorage继承自threading.local所以访问它的属性时每个线程得到的都是自己的副本。这就好像有一栋公寓楼self.thread_local所有线程都能看到这栋楼但每个线程只能进入自己的房间自己的属性副本进错房间是不可能的5.4 与DataProc对象的对比一个有趣的问题是data_dict、path、metal、filename这些都放在了TLS里那DataProc对象呢它的属性如interval_counts、total_segments等也是线程私有的吗让我们看一下DataProc的创建方式self.DataProcDataProcess()这段代码来自f:\grainserver\SocketServ\TaskHandle.py:56。DataProc是SocketServTaskHandle的实例属性也就是所有线程共享的。那它的属性会不会也有线程安全问题答案是会但GrainServer的代码通过锁来保护了。还记得write_data_dict方法里的锁吗在锁内部代码从self.DataProc中读取各种统计结果。因为有锁的保护同一时间只有一个线程在操作DataProc所以不会有问题。但是这是一个潜在的风险点。如果未来有代码在锁外面也修改DataProc的属性就可能出现线程安全问题。更好的设计可能是把DataProc也放在TLS里或者让每个线程创建自己的DataProc实例。不过当前的设计在锁的保护下也是正确的而且实现更简单。六、总结多线程是后端服务提升并发能力的重要手段但它也是一把双刃剑——用得好可以大幅提升性能和响应性用不好则会引入各种诡异的bug。本文从GrainServer的多线程架构出发系统讲解了双服务器每连接一线程的两层多线程模型为什么不能用全局变量以及线程局部存储的原理threading.local()的实现机制和使用方法互斥锁的作用和临界区的概念死锁的成因与规避策略ThreadLocalStorage类的完整代码解读GrainServer在多线程处理上的设计思路可以总结为能用TLS就不用共享变量尽量把数据放在线程局部存储中从根源上避免竞争必须共享的就用锁保护对于不可避免的共享资源用互斥锁确保安全粗粒度锁优先在并发量不高的情况下用简单的粗粒度锁换取代码的简洁和可靠性简洁至上不追求理论上的最优解而是选择最容易理解、最不容易出错的方案这些设计原则对于大多数中小型后端服务来说都是适用的。多线程编程没有必要搞得过于复杂简单、清晰、正确才是最重要的。在接下来的文章中我们将探讨GrainServer的日志系统设计看看一个好的日志系统如何为服务的稳定运行保驾护航。
