螺纹.get_ident（）在运行pytest时在不同线程之间返回相同的ID

发布于 2025-02-11 02:25:17 字数 623 浏览 0 评论 0原文

我已经使用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。

原文

I have created a small test case using pytest to demonstrate the issue:

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

The test is failing because get_ident returns the same ID for different threads. But when I add time.sleep(1) in each thread, the test passes.
I'm not sure I understand why.

I'm using Python 3.9.0 and Pytest 7.1.2.

分享到QQ

分享到微博

如果你对这篇内容有疑问，欢迎到本站社区发帖提问参与讨论，获取更多帮助，或者扫码二维码加入 Web 技术交流群。

发布评论

需要登录才能够评论，你可以免费注册一个本站的账号。

彼岸花似海 2025-02-18 02:25:17

来自 documentation ：

返回当前线程的“线程标识符”。这是一个非零整数。它的价值没有直接的含义；它旨在用作魔法曲奇，例如用来索引特定于线程的数据字典。线程标识符可能会在线程退出并创建另一个线程时回收。

由于您的线程运行太快（没有time.sleep（1）），因此ID被回收。

您可以向每个线程提供名称。此 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)}')

From the documentation of get_ident:

Return the ‘thread identifier’ of the current thread. This is a nonzero integer. Its value has no direct meaning; it is intended as a magic cookie to be used e.g. to index a dictionary of thread-specific data. Thread identifiers may be recycled when a thread exits and another thread is created.

Since your threads are running too quickly (without the time.sleep(1)), the ids are being recycled.

You can provide a name to each thread. This name does not have to be unique, but you can use it in your test (or in your application, if you need something that is unique in a context):

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)}')

回复收藏 0 原文

~没有更多了~