我使用pytest创建了一个小测试用例来演示该问题:

from threading import Thread, get_ident

def test_threading():
    threads = []
    ids = []

    def append_id():
        # time.sleep(1)
        ids.append(get_ident())

    for _ in range(5):
        thread = Thread(target=append_id)
        threads.append(thread)
        thread.start()

    for thread in threads:
        thread.join()

    assert len(set(ids)) == 5

测试失败,因为get_ident为不同的线程返回相同的ID.但当我在每个线程中添加time.sleep(1)时,测试通过了.

我正在使用Python 3.9.0和Pytest 7.1.2.

推荐答案

get_identdocumentation开始:

返回当前线程的"线程标识符".这是一个非零整数.其价值没有直接意义;它是一种神奇的cookie,可用于索引特定于线程的数据字典.当一个线程退出并创建另一个线程时,可以回收线程标识符.

由于您的线程运行得太快(没有time.sleep(1)),因此ID正在被回收.

您可以 for each 线程提供name.这name不一定是唯一的,但您可以在测试中使用它(或者在应用程序中,如果您需要上下文中唯一的内容):

from threading import Thread, get_ident, current_thread

def test_threading():
    threads = []
    names = []
    ids = []

    def append_id():
        # time.sleep(1)
        ids.append(get_ident())
        names.append(current_thread().name)

    for i in range(5):
        thread = Thread(target=append_id, name=f'Thread {i}')
        threads.append(thread)
        thread.start()

    for thread in threads:
        thread.join()

    assert len(set(names)) == 5
    print(f'Names: {names}')
    print(f'Ids: {set(ids)}')

Python相关问答推荐

如果条件为真,则Groupby.mean()

acme错误-Veritas错误:模块收件箱没有属性linear_util'

修复mypy错误-赋值中的类型不兼容(表达式具有类型xxx,变量具有类型yyy)

cv2.matchTemplate函数匹配失败

计算每个IP的平均值

删除marplotlib条形图上的底边

为什么Django管理页面和我的页面的其他CSS文件和图片都找不到?'

Pandas Data Wrangling/Dataframe Assignment

基于行条件计算(pandas)

在极中解析带有数字和SI前缀的字符串

如何防止Pandas将索引标为周期?

使用__json__的 pyramid 在客户端返回意外格式

BeautifulSoup:超过24个字符(从a到z)的迭代失败:降低了首次深入了解数据集的复杂性:

使用SeleniumBase保存和加载Cookie时出现问题

如何在Airflow执行日期中保留日期并将时间转换为00:00

当HTTP 201响应包含 Big Data 的POST请求时,应该是什么?  

在matplotlib中重叠极 map 以创建径向龙卷风图

.awk文件可以使用子进程执行吗?

普洛特利express 发布的人口普查数据失败

将时间序列附加到数据帧