Python 标准库之 threading （线程并行）

date

Mar 24, 2020

slug

5589750164580352

status

Published

示例

返回活跃线程的数量


In [1]: import threading
In [2]: threading.active_count()
Out[2]: 22In [3]: len(threading.enumerate())
Out[3]: 22

返回活跃线程列表


In [1]: import threading
In [2]: threading.enumerate()
Out[2:
[<_MainThread(MainThread, started 3744)>,
 <HistorySavingThread(IPythonHistorySavingThread, started 9356)>,
 <Thread(ThreadPoolExecutor-0_0, started daemon 6552)>,
 <Thread(ThreadPoolExecutor-0_1, started daemon 9492)>,
 <Thread(ThreadPoolExecutor-0_2, started daemon 8740)>,
 <Thread(ThreadPoolExecutor-0_3, started daemon 4128)>,
 <Thread(ThreadPoolExecutor-0_4, started daemon 288)>,
 <Thread(ThreadPoolExecutor-0_5, started daemon 2804)>,
 <Thread(ThreadPoolExecutor-0_6, started daemon 5848)>,
 <Thread(ThreadPoolExecutor-0_7, started daemon 9608)>,
 <Thread(ThreadPoolExecutor-0_8, started daemon 6348)>,
 <Thread(ThreadPoolExecutor-0_9, started daemon 7156)>,
 <Thread(ThreadPoolExecutor-0_10, started daemon 4268)>,
 <Thread(ThreadPoolExecutor-0_11, started daemon 7364)>,
 <Thread(ThreadPoolExecutor-0_12, started daemon 7532)>,
 <Thread(ThreadPoolExecutor-0_13, started daemon 5888)>,
 <Thread(ThreadPoolExecutor-0_14, started daemon 364)>,
 <Thread(ThreadPoolExecutor-0_15, started daemon 5816)>,
 <Thread(ThreadPoolExecutor-0_16, started daemon 9636)>,
 <Thread(ThreadPoolExecutor-0_17, started daemon 6756)>,
 <Thread(ThreadPoolExecutor-0_18, started daemon 10080)>,
 <Thread(ThreadPoolExecutor-0_19, started daemon 7264)>]

创建一个线程


In [1]: import threading
In [2]: thread = threading.Thread()
In [3]: thread
Out[3]: <Thread(Thread-1, initial)>

为线程绑定任务


In [1]: import threading
In [2]: thread = threading.Thread()
In [3]: def f1():
   ...:     print("Her")
   ...:
In [4]: thread.target = f1
In [5]: thread.target
Out[5]: <function __main__.f1()>

或


In [1]: import threading
In [2]: def f1():
   ...:     print("Her")
   ...:
In [2]: thread = threading.Thread(target=f1)

为线程任务传递参数


In [1]: import threading
In [2]: def f1(i):
   ...:     print("Her"+str(i))
   ...:
In [2]: thread = threading.Thread(target=f1, args=(666,))

运行线程


In [1]: import threading
In [2]: def f1(i):
   ...:     print("Her"+str(i))
   ...:
In [2]: thread = threading.Thread(target=f1, args=(666,))
In [3]: thread.start()
Her666

注意：
In [1]: thread.run()
Her666
start() 方法是启动一个子线程
run() 方法并不启动一个新线程，而是在主线程中调用了任务函数。
start() 它在一个线程里最多只能被调用一次。

阻塞调用的线程（守护线程）

不使用 join() ：


import threading
import time
def f1():
    for i in range(2):
        print("thread1", time.ctime())
        time.sleep(1)
thread1 = threading.Thread(target=f1)
thread1.start()
print("done")

输出：


thread1 Wed Mar 18 18:52:23 2020
done
thread1 Wed Mar 18 18:52:24 2020

使用 join() ：


import threading
import time
def f1():
    for i in range(2):
        print("thread1", time.ctime())
        time.sleep(1)
thread1 = threading.Thread(target=f1)
thread1.start()
thread1.join()
print("done")

输出：


thread1Wed Mar 18 18:55:55 2020
thread1Wed Mar 18 18:55:56 2020
done

使用 join() 会将调用这个线程的主线程阻塞，等待当前线程终止后才将控制权交还给主线程。

多线程访问临界值


import threading
import time
a = 0def f1():
    global a
    t = a + 1    print(t)
    time.sleep(1) # 模拟修改变量值的时间过程    a = t
threads = [threading.Thread(target=f1) for _ in range(5)]
for _ in threads:
    _.start()

输出：

CPU 太快了，导致 time.sleep(1) 休眠没有结束就跑完所有线程，此时线程获取到 a 的值都是 0

线程锁


# 创建线程锁lock = threading.Lock()
# 获得锁lock.acquire()
# 释放锁lock.release()

为临界资源加锁：


import threading
import time
lock = threading.Lock()
a = 0def f1():
    global a
    lock.acquire()
    t = a + 1    print(t)
    time.sleep(1) # 模拟修改变量值的时间过程    a = t
    lock.release()
threads = [threading.Thread(target=f1) for _ in range(5)]
for _ in threads:
    _.start()

输出：

参考

threading.active_count() ：返回当前活跃的线程对象的数量 threading.current_thread() ：返回当前线程对象 threading.excepthook(args, /) ：处理 Thread.run() 未捕获的异常 threading.get_ident() ：返回当前线程的索 threading.get_native_id() ：返回内核分配给当前线程的原生集成线程 ID threading.enumerate() ：以列表形式返回当前所有存活的 Thread 对象 threading.main_thread() ：返回主 Thread 对象 threading.settrace(func) ：为所有 threading 模块开始的线程设置追踪函数 threading.setprofile(func) ：为所有 threading 模块开始的线程设置性能测试函数 threading.stack_size([size]) ：返回创建线程时用的堆栈大小 threading.TIMEOUT_MAX ：阻塞函数（ Lock.acquire(), RLock.acquire(), Condition.wait(), …）中形参 timeout 允许的最大值

threading.local() ：创建本地线程数据

线程对象

threading.Thread(group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None) ：创建线程对象

参数	说明
group	group 应该为 `None`；为了日后扩展 `ThreadGroup` 类实现而保留
target	用于 `run()`方法调用的可调用对象
name	线程名称
args	调用目标函数的参数元组
kwargs	用目标函数的关键字参数字典
daemon	是否为守护模式，默认值 `None` 为线程将继承当前线程的守护模式属性

start() ：开始线程活动。 run() ：代表线程活动的方法。标准的 run() 方法会对作为 target 参数传递给该对象构造器的可调用对象（如果存在）发起调用，并附带从 args 和 kwargs 参数分别获取的位置和关键字参数 join(timeout=None) ：等待，直到线程终结。 name ：只用于识别的字符串。 getName()，setName() ：获取/设置 name 的值。 ident ：这个线程的 ‘线程标识符’ native_id ：该线程的当前线程ID。 is_alive() ：返回线程是否存活。 daemon ：一个表示这个线程是（True）否（False）守护线程的布尔值。 isDaemon()，setDaemon() ：获取/设置 name的值。

锁对象

threading.Lock() ：创建所。 acquire(blocking=True, timeout=-1) ：可以阻塞或非阻塞地获得锁。 release() ：释放一个锁。 locked() ：如果获得了锁则返回真值。

更多参考