ZetCode

Python time.thread_time_ns 函数

上次修改时间:2025 年 4 月 11 日

本综合指南探讨了 Python 的 time.thread_time_ns 函数,该函数以纳秒为单位返回线程特定的 CPU 时间。我们将介绍精确计时、性能测量和实际示例。

基本定义

time.thread_time_ns 函数以纳秒为单位返回当前线程的系统和用户 CPU 时间总和。它是线程特定的,不包括睡眠期间花费的时间。

主要特征:纳秒精度、线程特定测量以及用于分析 CPU 密集型操作。 返回值是一个整数,以实现最大精度。

基本线程时间测量

time.thread_time_ns 最简单的用法是测量当前线程的 CPU 时间。 此示例显示了基本用法以及转换为秒。

basic_thread_time.py
import time

# Get current thread CPU time in nanoseconds
thread_time = time.thread_time_ns()
print(f"Thread CPU time: {thread_time} ns")

# Convert to seconds
seconds = thread_time / 1e9
print(f"Thread CPU time: {seconds:.9f} seconds")

此示例演示了如何获取线程特定的 CPU 时间并将其转换为秒。 纳秒精度可以实现极其精确的计时。

请注意,这仅测量当前线程使用的 CPU 时间,而不是挂钟时间或在其他线程中花费的时间。

测量 CPU 密集型操作

time.thread_time_ns 非常适合测量 CPU 密集型操作。 此示例显示了如何计时计算密集型任务。

cpu_bound.py
import time

def compute_factorial(n):
    result = 1
    for i in range(1, n+1):
        result *= i
    return result

# Start timer
start = time.thread_time_ns()

# Execute CPU-bound operation
fact = compute_factorial(10000)

# Calculate duration
end = time.thread_time_ns()
duration_ns = end - start

print(f"Computation took {duration_ns} ns")
print(f"Computation took {duration_ns / 1e6:.3f} ms")

此模式准确地仅测量计算中花费的 CPU 时间,不包括任何等待或睡眠时间。 结果以纳秒为单位。

该示例还显示了转换为毫秒以获得更易读的输出。

比较线程时间和进程时间

此示例将 thread_time_nsprocess_time_ns 进行比较,以显示线程特定与进程范围的 CPU 时间测量。

thread_vs_process.py
import time
import threading

def cpu_work():
    sum(range(10**6))

# Single thread measurement
thread_start = time.thread_time_ns()
process_start = time.process_time_ns()
cpu_work()
thread_end = time.thread_time_ns()
process_end = time.process_time_ns()

print(f"Single thread:")
print(f"  Thread time: {(thread_end - thread_start) / 1e6:.3f} ms")
print(f"  Process time: {(process_end - process_start) / 1e6:.3f} ms")

# Multi-thread measurement
def worker():
    thread_start = time.thread_time_ns()
    cpu_work()
    thread_end = time.thread_time_ns()
    print(f"  Worker thread time: {(thread_end - thread_start) / 1e6:.3f} ms")

print("\nMulti-thread:")
process_start = time.process_time_ns()
threads = [threading.Thread(target=worker) for _ in range(2)]
for t in threads:
    t.start()
for t in threads:
    t.join()
process_end = time.process_time_ns()

print(f"  Total process time: {(process_end - process_start) / 1e6:.3f} ms")

该示例表明 thread_time_ns 仅测量当前线程的 CPU 时间,而 process_time_ns 包括进程中的所有线程。

在多线程场景中,thread_time_ns 提供精确的每个线程指标。

分析函数执行

此示例创建一个装饰器,该装饰器使用 thread_time_ns 分析函数执行,以进行精确的 CPU 时间测量。

profiler.py
import time

def profile(func):
    def wrapper(*args, **kwargs):
        start = time.thread_time_ns()
        result = func(*args, **kwargs)
        end = time.thread_time_ns()
        duration_ns = end - start
        print(f"{func.__name__} took {duration_ns} ns ({duration_ns / 1e6:.3f} ms)")
        return result
    return wrapper

@profile
def calculate_primes(n):
    primes = []
    candidate = 2
    while len(primes) < n:
        is_prime = True
        for p in primes:
            if p * p > candidate:
                break
            if candidate % p == 0:
                is_prime = False
                break
        if is_prime:
            primes.append(candidate)
        candidate += 1
    return primes

calculate_primes(1000)

分析器装饰器仅测量执行函数所花费的 CPU 时间,不包括在 I/O 或睡眠中花费的任何时间。 这对于 CPU 密集型任务来说是理想的。

纳秒精度甚至可以检测到非常小的性能差异。

比较 CPU 密集型操作与 I/O 密集型操作

此示例演示了 thread_time_ns 在 CPU 密集型和 I/O 密集型操作中的不同行为。

cpu_vs_io.py
import time

def cpu_task():
    sum(i*i for i in range(10**6))

def io_task():
    time.sleep(0.1)  # Simulate I/O wait

print("CPU-bound task:")
start = time.thread_time_ns()
cpu_task()
end = time.thread_time_ns()
print(f"  Thread CPU time: {(end - start) / 1e6:.3f} ms")

print("\nI/O-bound task:")
start = time.thread_time_ns()
io_task()
end = time.thread_time_ns()
print(f"  Thread CPU time: {(end - start) / 1e6:.3f} ms")

CPU 密集型任务显示显着的线程 CPU 时间,而 I/O 密集型任务尽管花费了挂钟时间,但显示的 CPU 时间几乎为零。

这表明 thread_time_ns 仅测量实际的 CPU 使用率,而不是等待 I/O 操作所花费的时间。

多线程性能分析

此示例使用 thread_time_ns 分析多线程应用程序中的性能,显示每个线程的 CPU 使用率。

multi_thread.py
import time
import threading

def worker(worker_id):
    start = time.thread_time_ns()
    
    # Simulate work with different CPU intensities
    for i in range(worker_id * 1000000):
        x = i * i
    
    end = time.thread_time_ns()
    duration = (end - start) / 1e6
    print(f"Worker {worker_id} CPU time: {duration:.3f} ms")

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

for t in threads:
    t.join()

每个线程报告自己的 CPU 时间,从而可以分析工作如何在线程之间分配。 较重的工作负载显示成比例的较高 CPU 时间。

此技术对于识别并行处理应用程序中的线程不平衡非常有价值。

小代码段的微基准测试

thread_time_ns 的纳秒精度使其成为小代码段微基准测试的理想选择。 此示例比较了两种方法。

microbenchmark.py
import time

def test_list_comprehension():
    start = time.thread_time_ns()
    [x*x for x in range(1000)]
    end = time.thread_time_ns()
    return end - start

def test_loop():
    start = time.thread_time_ns()
    result = []
    for x in range(1000):
        result.append(x*x)
    end = time.thread_time_ns()
    return end - start

# Warm-up runs (JIT compilation, cache effects)
test_list_comprehension()
test_loop()

# Actual measurements
lc_times = [test_list_comprehension() for _ in range(100)]
loop_times = [test_loop() for _ in range(100)]

avg_lc = sum(lc_times) / len(lc_times)
avg_loop = sum(loop_times) / len(loop_times)

print(f"List comprehension avg: {avg_lc:.1f} ns")
print(f"Traditional loop avg: {avg_loop:.1f} ns")
print(f"Difference: {avg_loop - avg_lc:.1f} ns ({avg_loop/avg_lc:.1f}x)")

该示例显示了如何通过使用多次运行并进行平均来正确地对小代码差异进行微基准测试。 预热运行有助于避免测量偏差。

纳秒精度揭示了看似相似的实现之间的细微性能差异。

最佳实践

资料来源

作者

我叫 Jan Bodnar,是一位充满热情的程序员,拥有丰富的编程经验。 我从 2007 年开始撰写编程文章。到目前为止,我撰写了超过 1,400 篇文章和 8 本电子书。 我拥有超过十年的编程教学经验。

列出所有 Python 教程