python - 如何在Python中测试线程

Question

我正在学习如何在 Python 中测试线程，因为我以前从未这样做过。我已经加了锁，但它说我有一个断言错误，这很好。我想知道下面的代码是否正确

import threading

i = 0

def test():
    global i
    for _ in range(100000):
        with threading.Lock():
            i += 1

threads = [threading.Thread(target=test) for t in range(10)]
for t in threads:
    t.start()

for t in threads:
    t.join()

assert i == 1000000, i

Answer 1

您的问题是您在每次迭代中创建一个新的锁，该锁始终处于解锁状态。

这样就可以了，因为您的线程将尝试获取相同的锁。

import threading

i = 0

lock = threading.Lock()

def test():
    global i
    for _ in range(100000):
        with lock:
            i += 1

threads = [threading.Thread(target=test) for t in range(10)]
for t in threads:
    t.start()

for t in threads:
    t.join()

print(i)