python基础之线程

简单示例

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.")

线程间通信

1. 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()

2. 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('执行结束')

3. 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]