强制终止线程

https://blog.csdn.net/A18373279153/article/details/80050177?utm_medium=distribute.pc_relevant.none-task-blog-BlogCommendFromMachineLearnPai2-2.nonecase&depth_1-utm_source=distribute.pc_relevant.none-task-blog-BlogCommendFromMachineLearnPai2-2.nonecase

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

import threading
import time
import inspect
import ctypes


def _async_raise(tid, exctype):
    """raises the exception, performs cleanup if needed"""
    tid = ctypes.c_long(tid)
    if not inspect.isclass(exctype):
        exctype = type(exctype)
    res = ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, ctypes.py_object(exctype))
    if res == 0:
        raise ValueError("invalid thread id")
    elif res != 1:
        # """if it returns a number greater than one, you're in trouble,
        # and you should call it again with exc=NULL to revert the effect"""
        ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, None)
        raise SystemError("PyThreadState_SetAsyncExc failed")


def stop_thread(thread):
    _async_raise(thread.ident, SystemExit)


def test():
    while True:
        print('-------')
        # time.sleep(0.5)
        pass


if __name__ == "__main__":
    t = threading.Thread(target=test)
    t.start()
    time.sleep(5.2)
    print("main thread sleep finish")
    stop_thread(t)

获取返回值

https://blog.csdn.net/it_is_arlon/article/details/86594416

  • 实现逻辑：在 Thread object 中存储与取出值

    • 注意：这种方法不能在多进程，Process 中实现

      • 原因：主进程和子进程中的 Process oject 不是同一个
      • 而 Thread oject 在主线程和子线程中是同一个（id 相同）
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

import threading


class ResultThread(threading.Thread):
    def __init__(self, target=None, args=(), kwargs={}):
        super().__init__()
        self.target = target
        self.args = args
        self.kwargs = kwargs

    def run(self):
        self.result = self.target(*self.args, **self.kwargs)

    def get_result(self):
        try:
        return self.result
        except Exception:
        return None


def timeout_thread(timeout, target, *args, **kwargs):
    job = ResultThread(target=target, args=args, kwargs=kwargs)
    job.start()
    job.join(timeout)

    if job.is_alive():
        stop_thread(job)
    return job.get_result()

定时任务

使用 threading.Timer 类

  • Timer 类是 Thread 类的子类

    • Timer(delay, target, args, kwargs)

FIFO 先进先出的线程锁

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

import threading
from queue import Queue

class FIFOLock:
    def __init__(self):
        self._queue = Queue()
        self._lock = threading.Lock()
    
    def acquire(self, blocking=True):
        # 创建一个事件用于通知当前线程
        event = threading.Event()
        
        with self._lock:
            # 将事件放入队列
            self._queue.put(event)
            # 检查是否是队首
            is_first = self._queue.queue[0] is event
        
        if is_first:
            # 如果是第一个，直接获得锁
            event.set()
        
        # 等待轮到自己
        if blocking:
            event.wait()
            return True
        else:
            return event.is_set()
    
    def release(self):
        with self._lock:
            # 移除当前线程的事件
            if not self._queue.empty():
                self._queue.get()
            
            # 通知下一个等待的线程
            if not self._queue.empty():
                next_event = self._queue.queue[0]
                next_event.set()
    
    def __enter__(self):
        self.acquire()
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.release()
        return False

支持并行量限制的 FIFO 线程锁

不允许 reentrant 版本

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256

import time
import uuid
import threading
from queue import Queue

from loguru import logger


class ThreadId(threading.local):
    """线程本地存储，每个线程有自己独立的id"""

    id: str | None = None


thread_id = ThreadId()


class FIFOThreadLock:
    def __init__(self, max_concurrent=1):
        """
        初始化FIFO线程锁

        Args:
            max_concurrent: 最大并行执行的线程数量,默认为1(相当于普通锁)
        """
        self._queue = Queue()
        self._lock = threading.Lock()
        self._max_concurrent = max_concurrent
        self._current_count = 0
        # 记录哪些线程已经获得了锁(用于防止重复release)
        self._acquired_threads = set()

    def _get_thread_id(self):
        """获取当前线程的唯一标识符"""
        if not thread_id.id:
            # ✅ 修复：正确地设置ThreadLocal对象的id属性
            thread_id.id = str(uuid.uuid4())
        return thread_id.id
        # return threading.get_ident()

    def acquire(self, blocking=True):
        """获取锁"""
        tid = self._get_thread_id()  # ✅ 使用辅助方法获取线程ID

        # 创建一个事件用于通知当前线程
        event = threading.Event()

        with self._lock:
            # 将事件和线程ID关联放入队列
            self._queue.put((event, tid))
            # 检查是否可以立即执行
            can_proceed = self._current_count < self._max_concurrent
            if can_proceed:
                self._current_count += 1
                self._acquired_threads.add(tid)
                event.set()

        # 等待轮到自己
        if blocking:
            event.wait()
            # 如果事件被设置,但当前线程不在acquired_threads中
            # 说明是在release()中被唤醒的,需要添加到集合中
            with self._lock:
                if tid not in self._acquired_threads:
                    self._acquired_threads.add(tid)
            return True
        else:
            # 非阻塞模式
            is_acquired = event.is_set()
            if not is_acquired:
                # 如果没有获得锁,需要从队列中移除并清理
                with self._lock:
                    # 需要遍历队列找到对应的事件并移除
                    temp_items = []
                    found = False
                    while not self._queue.empty():
                        item = self._queue.get()
                        if item[0] is event and not found:
                            found = True
                            continue
                        temp_items.append(item)
                    # 将其他项放回队列
                    for item in temp_items:
                        self._queue.put(item)
            return is_acquired

    def release(self):
        """释放锁"""
        tid = self._get_thread_id()  # ✅ 使用辅助方法获取线程ID（保持一致性）

        with self._lock:
            # 检查当前线程是否真的持有锁
            if tid not in self._acquired_threads:
                # 防止重复release或未acquire就release
                raise RuntimeError(f"线程 {tid} 试图释放未持有的锁")

            # 移除当前线程的标记
            self._acquired_threads.remove(tid)

            # 从队列中找到并移除当前线程的事件
            found = False
            temp_items = []
            while not self._queue.empty():
                item = self._queue.get()
                if item[1] == tid and not found:
                    # 找到了当前线程,不放回队列
                    found = True
                    logger.debug(f"[DEBUG] 线程 {tid} 从队列中移除")
                else:
                    temp_items.append(item)

            # 将其他项放回队列
            for item in temp_items:
                self._queue.put(item)

            if not found:
                logger.debug(f"[WARNING] 线程 {tid} 在队列中未找到!")

            # 减少当前计数(使用max确保不会变成负数)
            self._current_count = max(0, self._current_count - 1)
            logger.debug(
                f"[DEBUG] 释放后计数: {self._current_count}, 队列大小: {self._queue.qsize()}"
            )

            # 通知下一个等待的线程(如果有空位)
            while (
                not self._queue.empty() and self._current_count < self._max_concurrent
            ):
                next_item = self._queue.queue[0]
                next_event, next_thread_id = next_item
                if not next_event.is_set():
                    self._current_count += 1
                    logger.debug(
                        f"[DEBUG] 唤醒线程 {next_thread_id}, 新计数: {self._current_count}"
                    )
                    # 注意:这里先设置事件,线程会在acquire()中添加自己到_acquired_threads
                    next_event.set()
                    break
                else:
                    # 如果事件已经设置,从队列中移除
                    logger.debug(f"[DEBUG] 跳过已设置的事件 {next_thread_id}")
                    self._queue.get()
                    break

    def __enter__(self):
        self.acquire()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.release()
        return False


# ============ 测试代码 ============


def test_basic_usage():
    """测试基本功能"""
    logger.debug("=== 测试1: 基本功能（3并发） ===\n")

    fifo_lock = FIFOThreadLock(max_concurrent=3)

    def worker(thread_id):
        logger.debug(f"线程 {thread_id} 正在等待")
        with fifo_lock:
            logger.debug(f"✓ 线程 {thread_id} 获得了锁,开始执行")
            time.sleep(1)
            logger.debug(f"✓ 线程 {thread_id} 完成工作")

    threads = []
    for i in range(5):
        t = threading.Thread(target=worker, args=(i,))
        threads.append(t)
        t.start()
        time.sleep(0.1)

    for t in threads:
        t.join()

    logger.debug(f"\n最终计数: {fifo_lock._current_count} (应该为0)")
    logger.debug(f"持有锁的线程: {fifo_lock._acquired_threads} (应该为空)\n")


def test_reacquire():
    """测试线程多次获取和释放锁（修复的关键测试）"""
    logger.debug("=== 测试2: 线程多次获取释放锁 ===\n")

    fifo_lock = FIFOThreadLock(max_concurrent=2)

    def worker(worker_id, iterations):
        for i in range(iterations):
            logger.debug(f"Worker {worker_id} - 第{i+1}次尝试获取锁")
            with fifo_lock:
                logger.debug(f"✓ Worker {worker_id} - 第{i+1}次成功获取锁")
                time.sleep(0.3)
                logger.debug(f"✓ Worker {worker_id} - 第{i+1}次释放锁")

    threads = []
    for i in range(3):
        t = threading.Thread(target=worker, args=(i, 3))
        threads.append(t)
        t.start()

    for t in threads:
        t.join()

    logger.debug(f"\n最终计数: {fifo_lock._current_count} (应该为0)")
    logger.debug(f"持有锁的线程: {fifo_lock._acquired_threads} (应该为空)\n")


def test_exception_handling():
    """测试异常处理"""
    logger.debug("=== 测试3: 异常处理 ===\n")

    fifo_lock = FIFOThreadLock(max_concurrent=2)

    def bad_worker(thread_id):
        logger.debug(f"线程 {thread_id} (会抛异常) 正在等待")
        try:
            with fifo_lock:
                logger.debug(f"线程 {thread_id} 获得了锁,即将抛出异常")
                raise ValueError("模拟异常")
        except ValueError:
            logger.debug(f"✓ 线程 {thread_id} 捕获了异常,锁已自动释放")

    def normal_worker(thread_id):
        logger.debug(f"线程 {thread_id} (正常) 正在等待")
        with fifo_lock:
            logger.debug(f"✓ 线程 {thread_id} 获得了锁")
            time.sleep(0.5)
            logger.debug(f"✓ 线程 {thread_id} 完成工作")

    threads = []
    for i in range(5):
        if i == 2:
            t = threading.Thread(target=bad_worker, args=(i,))
        else:
            t = threading.Thread(target=normal_worker, args=(i,))
        threads.append(t)
        t.start()
        time.sleep(0.1)

    for t in threads:
        t.join()

    logger.debug(f"\n最终计数: {fifo_lock._current_count} (应该为0)")
    logger.debug(f"持有锁的线程: {fifo_lock._acquired_threads} (应该为空)\n")


if __name__ == "__main__":
    test_basic_usage()
    logger.debug("\n" + "=" * 60 + "\n")
    test_reacquire()
    logger.debug("\n" + "=" * 60 + "\n")
    test_exception_handling()
    logger.debug("\n所有测试完成!")

允许 reentrant 版本

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321

import time
import uuid
import threading
from queue import Queue

from loguru import logger


class ThreadId(threading.local):
    """线程本地存储，每个线程有自己独立的id"""

    id: str | None = None


thread_id = ThreadId()


class FIFOThreadLock:
    def __init__(self, max_concurrent=1):
        """
        初始化FIFO线程锁（支持可重入）

        Args:
            max_concurrent: 最大并行执行的线程数量,默认为1(相当于普通锁)
        """
        self._queue = Queue()
        self._lock = threading.Lock()
        self._max_concurrent = max_concurrent
        self._current_count = 0
        # 记录哪些线程已经获得了锁(用于防止重复release)
        self._acquired_threads = set()
        # ✅ 新增：记录每个线程的重入次数
        self._reentrant_count = {}

    def _get_thread_id(self):
        """获取当前线程的唯一标识符"""
        if not thread_id.id:
            thread_id.id = str(uuid.uuid4())
        return thread_id.id

    def acquire(self, blocking=True):
        """获取锁（支持可重入）"""
        tid = self._get_thread_id()

        # ✅ 检查是否是重入调用
        with self._lock:
            if tid in self._acquired_threads:
                # 同一线程重入，只增加计数
                self._reentrant_count[tid] = self._reentrant_count.get(tid, 1) + 1
                logger.debug(f"线程 {tid} 重入，当前重入次数: {self._reentrant_count[tid]}")
                return True

        # 创建一个事件用于通知当前线程
        event = threading.Event()

        with self._lock:
            # 将事件和线程ID关联放入队列
            self._queue.put((event, tid))
            # 检查是否可以立即执行
            can_proceed = self._current_count < self._max_concurrent
            if can_proceed:
                self._current_count += 1
                self._acquired_threads.add(tid)
                self._reentrant_count[tid] = 1  # ✅ 初始化重入计数
                event.set()

        # 等待轮到自己
        if blocking:
            event.wait()
            # 如果事件被设置,但当前线程不在acquired_threads中
            # 说明是在release()中被唤醒的,需要添加到集合中
            with self._lock:
                if tid not in self._acquired_threads:
                    self._acquired_threads.add(tid)
                    self._reentrant_count[tid] = 1  # ✅ 初始化重入计数
            return True
        else:
            # 非阻塞模式
            is_acquired = event.is_set()
            if not is_acquired:
                # 如果没有获得锁,需要从队列中移除并清理
                with self._lock:
                    # 需要遍历队列找到对应的事件并移除
                    temp_items = []
                    found = False
                    while not self._queue.empty():
                        item = self._queue.get()
                        if item[0] is event and not found:
                            found = True
                            continue
                        temp_items.append(item)
                    # 将其他项放回队列
                    for item in temp_items:
                        self._queue.put(item)
            else:
                # ✅ 非阻塞模式成功获取锁，初始化重入计数
                with self._lock:
                    if tid not in self._reentrant_count:
                        self._reentrant_count[tid] = 1
            return is_acquired

    def release(self):
        """释放锁（支持可重入）"""
        tid = self._get_thread_id()

        with self._lock:
            # 检查当前线程是否真的持有锁
            if tid not in self._acquired_threads:
                raise RuntimeError(f"线程 {tid} 试图释放未持有的锁")

            # ✅ 检查重入次数
            reentrant_count = self._reentrant_count.get(tid, 1)
            if reentrant_count > 1:
                # 还有重入层级，只减少计数，不真正释放
                self._reentrant_count[tid] = reentrant_count - 1
                logger.debug(f"线程 {tid} 减少重入计数，剩余: {self._reentrant_count[tid]}")
                return

            # ✅ 重入计数为1，执行真正的释放
            # 移除当前线程的标记
            self._acquired_threads.remove(tid)
            del self._reentrant_count[tid]  # ✅ 清理重入计数

            # 从队列中找到并移除当前线程的事件
            found = False
            temp_items = []
            while not self._queue.empty():
                item = self._queue.get()
                if item[1] == tid and not found:
                    # 找到了当前线程,不放回队列
                    found = True
                    logger.debug(f"[DEBUG] 线程 {tid} 从队列中移除")
                else:
                    temp_items.append(item)

            # 将其他项放回队列
            for item in temp_items:
                self._queue.put(item)

            if not found:
                logger.debug(f"[WARNING] 线程 {tid} 在队列中未找到!")

            # 减少当前计数(使用max确保不会变成负数)
            self._current_count = max(0, self._current_count - 1)
            logger.debug(
                f"[DEBUG] 释放后计数: {self._current_count}, 队列大小: {self._queue.qsize()}"
            )

            # 通知下一个等待的线程(如果有空位)
            while (
                not self._queue.empty() and self._current_count < self._max_concurrent
            ):
                next_item = self._queue.queue[0]
                next_event, next_thread_id = next_item
                if not next_event.is_set():
                    self._current_count += 1
                    logger.debug(
                        f"[DEBUG] 唤醒线程 {next_thread_id}, 新计数: {self._current_count}"
                    )
                    next_event.set()
                    break
                else:
                    # 如果事件已经设置,从队列中移除
                    logger.debug(f"[DEBUG] 跳过已设置的事件 {next_thread_id}")
                    self._queue.get()
                    break

    def __enter__(self):
        self.acquire()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.release()
        return False


# ============ 测试代码 ============


def test_basic_usage():
    """测试基本功能"""
    logger.info("=== 测试1: 基本功能（3并发） ===\n")

    fifo_lock = FIFOThreadLock(max_concurrent=3)

    def worker(thread_id):
        logger.info(f"线程 {thread_id} 正在等待")
        with fifo_lock:
            logger.info(f"✓ 线程 {thread_id} 获得了锁,开始执行")
            time.sleep(1)
            logger.info(f"✓ 线程 {thread_id} 完成工作")

    threads = []
    for i in range(5):
        t = threading.Thread(target=worker, args=(i,))
        threads.append(t)
        t.start()
        time.sleep(0.1)

    for t in threads:
        t.join()

    logger.info(f"\n最终计数: {fifo_lock._current_count} (应该为0)")
    logger.info(f"持有锁的线程: {fifo_lock._acquired_threads} (应该为空)\n")


def test_reentrant():
    """✅ 测试可重入功能（修复的关键）"""
    logger.info("=== 测试2: 可重入功能 ===\n")

    fifo_lock = FIFOThreadLock(max_concurrent=2)

    def nested_worker(worker_id):
        logger.info(f"Worker {worker_id} - 第一层获取锁")
        with fifo_lock:
            logger.info(f"✓ Worker {worker_id} - 第一层成功")
            time.sleep(0.2)
            
            logger.info(f"Worker {worker_id} - 第二层获取锁（嵌套）")
            with fifo_lock:
                logger.info(f"✓ Worker {worker_id} - 第二层成功（可重入）")
                time.sleep(0.2)
                
                logger.info(f"Worker {worker_id} - 第三层获取锁（嵌套）")
                with fifo_lock:
                    logger.info(f"✓ Worker {worker_id} - 第三层成功（可重入）")
                    time.sleep(0.2)
                logger.info(f"✓ Worker {worker_id} - 第三层释放")
            logger.info(f"✓ Worker {worker_id} - 第二层释放")
        logger.info(f"✓ Worker {worker_id} - 第一层释放")

    threads = []
    for i in range(3):
        t = threading.Thread(target=nested_worker, args=(i,))
        threads.append(t)
        t.start()
        time.sleep(0.05)

    for t in threads:
        t.join()

    logger.info(f"\n最终计数: {fifo_lock._current_count} (应该为0)")
    logger.info(f"持有锁的线程: {fifo_lock._acquired_threads} (应该为空)")
    logger.info(f"重入计数: {fifo_lock._reentrant_count} (应该为空)\n")


def test_multiple_acquire():
    """测试线程多次获取和释放锁"""
    logger.info("=== 测试3: 线程多次获取释放锁 ===\n")

    fifo_lock = FIFOThreadLock(max_concurrent=2)

    def worker(worker_id, iterations):
        for i in range(iterations):
            logger.info(f"Worker {worker_id} - 第{i+1}次尝试获取锁")
            with fifo_lock:
                logger.info(f"✓ Worker {worker_id} - 第{i+1}次成功获取锁")
                time.sleep(0.3)
                logger.info(f"✓ Worker {worker_id} - 第{i+1}次释放锁")

    threads = []
    for i in range(3):
        t = threading.Thread(target=worker, args=(i, 3))
        threads.append(t)
        t.start()

    for t in threads:
        t.join()

    logger.info(f"\n最终计数: {fifo_lock._current_count} (应该为0)")
    logger.info(f"持有锁的线程: {fifo_lock._acquired_threads} (应该为空)\n")


def test_exception_handling():
    """测试异常处理"""
    logger.info("=== 测试4: 异常处理 ===\n")

    fifo_lock = FIFOThreadLock(max_concurrent=2)

    def bad_worker(thread_id):
        logger.info(f"线程 {thread_id} (会抛异常) 正在等待")
        try:
            with fifo_lock:
                logger.info(f"线程 {thread_id} 获得了锁,即将抛出异常")
                raise ValueError("模拟异常")
        except ValueError:
            logger.info(f"✓ 线程 {thread_id} 捕获了异常,锁已自动释放")

    def normal_worker(thread_id):
        logger.info(f"线程 {thread_id} (正常) 正在等待")
        with fifo_lock:
            logger.info(f"✓ 线程 {thread_id} 获得了锁")
            time.sleep(0.5)
            logger.info(f"✓ 线程 {thread_id} 完成工作")

    threads = []
    for i in range(5):
        if i == 2:
            t = threading.Thread(target=bad_worker, args=(i,))
        else:
            t = threading.Thread(target=normal_worker, args=(i,))
        threads.append(t)
        t.start()
        time.sleep(0.1)

    for t in threads:
        t.join()

    logger.info(f"\n最终计数: {fifo_lock._current_count} (应该为0)")
    logger.info(f"持有锁的线程: {fifo_lock._acquired_threads} (应该为空)\n")


if __name__ == "__main__":
    test_basic_usage()
    logger.info("\n" + "=" * 60 + "\n")
    test_reentrant()
    logger.info("\n" + "=" * 60 + "\n")
    test_multiple_acquire()
    logger.info("\n" + "=" * 60 + "\n")
    test_exception_handling()
    logger.info("\n所有测试完成!")

threading.Condition

锁的获取和释放流程

参考：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

# Thread 1 (通知者)
cond.acquire()           # 1. 获取锁
# ... 修改共享状态 ...
cond.notify()            # 2. 发送通知(此时等待线程还在等待)
cond.release()           # 3. 释放锁

# Thread 2 (等待者)
cond.acquire()           # 4. 获取锁
while not condition:     # 5. 检查条件
    cond.wait()          # 6. 释放锁并等待通知
                         # 7. 被通知后,自动重新获取锁
# ... 条件满足,继续执行 ...
cond.release()           # 8. 释放锁

condtion.wait() 的原子操作

  • 释放锁
  • 阻塞等待通知
  • 收到通知后,自动重新获取锁
  • 返回(继续执行)

condition.notify() 不会释放锁，只会通知其他线程