简单示例
import threading
import time
def worker(num):
time.sleep(1)
print(num)
return
# target参数是线程执行的函数， args是函数的参数，需要一个元组
for i in range(10):
t = threading.Thread(target=worker, args=(i,), name="t.%d" % i)
t.start()
通过继承Thread类来实现
import threading
import time
class MyThread(threading.Thread):
def __init__(self,num):
threading.Thread.__init__(self)
self.num = num
def run(self): # 启动线程就是调用线程的run方法
print("running on number:%s" %self.num)
time.sleep(2)
if __name__ == '__main__':
t1 = MyThread(1)
t2 = MyThread(2)
t1.start()
t2.start()
threading.RLock和threading.Lock
RLock允许在同一线程中被多次acquire。而Lock却不允许这种情况。 如果使用RLock，那么acquire和release必须成对出现，即调用了n次acquire，必须调用n次的release才能真正释放所占用的锁。
加锁的例子
import time
from threading import Thread, Lock
value = 0
lock = Lock()
def getlock():
global value
# 加锁是为了防止在同一时间有多个线程操作数据
# 使用with会自动加锁和释放锁（acquire和release)
with lock:
new_value = value + 1
time.sleep(1)
value = new_value
threads= []
for i in range(100):
t = Thread(target=getlock)
t.start()
threads.append(t)
for t in threads:
# 主线程等待子线程执行完,程序再退出
t.join()
print(value)
RLock
rlock = threading.RLock()
rlock.acquire()
rlock.acquire() # 在同一线程内，程序不会堵塞。
rlock.release()
rlock.release() # 有几个acquire就需要几个release
print("end.")
线程间通信
- threading.Event
Event定义了一个“Flag”，如果“Flag”的值为False，那么当程序执行wait方法时就会阻塞，如果“Flag”值为True，那么wait方法时便不再阻塞（wait默认为阻塞状态）
import threading
def do(event):
print('start')
# 阻塞线程，等待Event传递事件
event.wait()
print('execute')
event_obj = threading.Event()
for i in range(10):
t = threading.Thread(target=do, args=(event_obj,))
t.start()
inp = input('input:')
if inp == 'true':
event_obj.set()
再来看一个生产者消费者的例子
#!/usr/bin/env python3.6
import threading
import time
from random import randint
def product(event, l):
integer = randint(10, 100)
l.append(integer) # 往列表中添加内容
print("product", integer)
event.set() # 设置flag为True
print("set")
time.sleep(1)
def consumer(event, l):
try:
integer = l.pop() # 从列表中取出内容
except:
pass
print("cosumer", integer)
event.wait() # 检测event状态，如果为false则阻塞
print("clear")
threads = []
l = []
event = threading.Event()
p = threading.Thread(target=product, args=(event, l))
p.start()
threads.append(p)
c = threading.Thread(target=consumer, args=(event, l))
c.start()
threads.append(c)
for t in threads:
t.join()
- Semaphore
为了防止不同的线程同时对一个公用的资源操作，需要设置同时访问资源的线程数。信号量同步基于内部计数器，acquire()---> 计数器减1(同时访问资源的线程数), release() ---> 计数器加1， 当计数器为0时，调用acquire()，线程阻塞
import threading
import time
def f1(i,lock):
name = t.getName()
with lock:
print(name, "acquice")
time.sleep(1)
print(name, "release")
lock = threading.Semaphore(5)
for i in range(30):
t = threading.Thread(target=f1,args=(i,lock,))
t.start()
print('执行结束')
- Codition
Condition被称为条件变量，除了提供与Lock类似的acquire和release方法外，还提供了wait和notify方法。
import threading
import time
def consumer(cond):
with cond:
cond.wait() # 阻塞
print("consumer")
def producer(cond):
with cond:
print("producer")
cond.notify() # 通知cond释放
condition = threading.Condition()
c1 = threading.Thread(name="c1", target=consumer, args=(condition,))
p = threading.Thread(name="p", target=producer, args=(condition,))
c1.start()
time.sleep(2)
p.start()
线程池
在标准库中有一个线程池模块
>>> from multiprocessing.pool import ThreadPool
>>> pool = ThreadPool(5)
>>> result = pool.map(lambda x: x**2, range(10))
>>> print(result)
[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
