day9---多线程，线程锁，队列

Posted 2020-08-09

进程、线程
http://www.ruanyifeng.com/blog/2013/04/processes_and_threads.html

使用threading模块实现多线程编程[综述]
Python这门解释性语言也有专门的线程模型，Python虚拟机使用GIL（Global Interpreter Lock，全局解释器锁）来互斥线程对共享资源的访问，但暂时无法利用多处理器的优势。

        在Python中我们主要是通过thread和 threading这两个模块来实现的，其中Python的threading模块是对thread做了一些包装的，可以更加方便的被使用，所以我们使用 threading模块实现多线程编程。这篇文章我们主要来看看Python对多线程编程的支持。

        在语言层面，Python对多线程提供了很好的支持，可以方便地支持创建线程、互斥锁、信号量、同步等特性。下面就是官网上介绍threading模块的基本资料及功能：

实现模块
        thread：多线程的底层支持模块，一般不建议使用；

        threading：对thread进行了封装，将一些线程的操作对象化。

threading模块
        Thread 线程类，这是我们用的最多的一个类，你可以指定线程函数执行或者继承自它都可以实现子线程功能；

        Timer与Thread类似，但要等待一段时间后才开始运行；

        Lock 锁原语，这个我们可以对全局变量互斥时使用；

        RLock 可重入锁，使单线程可以再次获得已经获得的锁；

        Condition 条件变量，能让一个线程停下来，等待其他线程满足某个“条件”；

        Event 通用的条件变量。多个线程可以等待某个事件发生，在事件发生后，所有的线程都被激活；

        Semaphore为等待锁的线程提供一个类似“等候室”的结构；

        BoundedSemaphore 与semaphore类似，但不允许超过初始值；

        Queue：实现了多生产者（Producer）、多消费者（Consumer）的队列，支持锁原语，能够在多个线程之间提供很好的同步支持。

其中Thread类
        是你主要的线程类，可以创建进程实例。该类提供的函数包括：

        getName(self) 返回线程的名字

        isAlive(self) 布尔标志，表示这个线程是否还在运行中

        isDaemon(self) 返回线程的daemon标志

        join(self, timeout=None) 程序挂起，直到线程结束，如果给出timeout，则最多阻塞timeout秒

        run(self) 定义线程的功能函数

        setDaemon(self, daemonic) 把线程的daemon标志设为daemonic

        setName(self, name) 设置线程的名字

        start(self) 开始线程执行

其中Queue提供的类
        Queue队列

        LifoQueue后入先出（LIFO）队列

        PriorityQueue 优先队列

Python threading模块

#!/usr/bin/env python
#-*- coding:utf-8 -*-
# Author:DCC
import threading
import time
def run(n):
    print("task",n)
    time.sleep(2)
t1 = threading.Thread(target=run,args=("t1",))
t2 = threading.Thread(target=run,args=("t2",))
t1.start()
t2.start()
print(t1.getName())
print(t2.getName())

#!/usr/bin/env python
#-*- coding:utf-8 -*-
# Author:DCC
import threading
import time
class MyThread(threading.Thread):
    def __init__(self,n):
        super(MyThread,self).__init__()
        self.n = n
    def run(self):
        print("running task",self.n)
        time.sleep(3)
t1 = MyThread("t1")
t2 = MyThread("t2")
if __name__ == ‘__main__‘:
    t1.start()
    t2.start()
    print(t1.getName())
    print(t2.getName())

Join & Daemon

一般情况下 主线程是不等待子线程是否执行完成的，只是触发一下，就不管了。


1、join ()方法：主线程A中，创建了子线程B，并且在主线程A中调用了B.join()，那么，主线程A会在调用的地方等待，直到子线程B完成操作后，才可以接着往下执行，那么在调用这个线程时可以使用被调用线程的join方法。
原型：join([timeout])
里面的参数时可选的，代表线程运行的最大时间，即如果超过这个时间，不管这个此线程有没有执行完毕都会被回收，然后主线程或函数都会接着执行的。

import threading  
import time  
class MyThread(threading.Thread):  
        def __init__(self,id):  
                threading.Thread.__init__(self)  
                self.id = id  
        def run(self):  
                x = 0  
                time.sleep(10)  
                print self.id  
  
if __name__ == "__main__":  
        t1=MyThread(999)  
        t1.start()  
        for i in range(5):  
                print i  
#执行结果
0  
1  
2  
3  
4  
999

机器上运行时，4和999之间，有明显的停顿。解释：线程t1 start后，主线程并没有等线程t1运行结束后再执行，而是先把5次循环打印执行完毕（打印到4），然后sleep（10）后，线程t1把传进去的999才打印出来。
现在，我们把join()方法加进去(其他代码不变)，看看有什么不一样，例子：

import threading  
import time  
class MyThread(threading.Thread):  
        def __init__(self,id):  
                threading.Thread.__init__(self)  
                self.id = id  
        def run(self):  
                x = 0  
                time.sleep(10)  
                print self.id  
   
if __name__ == "__main__":  
        t1=MyThread(999)  
        t1.start()  
        t1.join()  
        for i in range(5):  
                print i 
#执行结果
999  
0  
1  
2  
3  
4

2、setDaemon()方法。主线程A中，创建了子线程B，并且在主线程A中调用了B.setDaemon(),这个的意思是，把主线程A设置为守护线程，这时候，要是主线程A执行结束了，就不管子线程B是否完成,一并和主线程A退出.这就是setDaemon方法的含义，这基本和join是相反的。此外，还有个要特别注意的：必须在start() 方法调用之前设置，如果不设置为守护线程，程序会被无限挂起。
例子：就是设置子线程随主线程的结束而结束：

#!/usr/bin/env python
#-*- coding:utf-8 -*-
# Author:DCC
import threading
import time
def run(n):
    print("task",n)
    time.sleep(2)
    print("thread done...",n)
start_time = time.time()
t_objs = [] #存线程实例
for i in range(50):
    t = threading.Thread(target=run,args=("t-%s"% i,))
    t.setDaemon(True) #把当前线程设置为守护线程
    t.start()
    t_objs.append(t) # 为了不阻塞后面线程的启动，不在这里join，先放到-个列表里
    # print(t.getName())
#time.sleep(2)
print(threading.active_count())
# for i in t_objs: #循环线程实例列表，等待所有线程执行完毕
#     t.join()
print(time.time()-start_time)

 线程锁

CPU执行任务时，在线程之间是进行随机调度的，并且每个线程可能只执行n条代码后就转而执行另外一条线程。由于在一个进程中的多个线程之间是共享资源和数据的，这就容易造成资源抢夺或脏数据，于是就有了锁的概念，限制某一时刻只有一个线程能访问某个指定的数据。

未枷锁

import threading
import time
NUM = 0
def show():
    global NUM
    NUM += 1
    name = t.getName()
    time.sleep(1)       # 注意，这行语句的位置很重要，必须在NUM被修改后，否则观察不到脏数据的现象。
    print(name, "执行完毕后，NUM的值为： ", NUM)
 
for i in range(10):
    t = threading.Thread(target=show)
    t.start()
print(‘main thread stop‘)

LOCK锁

普通锁，也叫互斥锁，是独占的，同一时刻只有一个线程被放行。

import time
import threading
NUM = 10
def func(lock):
    global NUM
    lock.acquire()  # 让锁开始起作用
    NUM -= 1
    time.sleep(1)
    print(NUM)
    lock.release()  # 释放锁
lock = threading.Lock()  # 实例化一个锁对象
for i in range(10):
    t = threading.Thread(target=func, args=(lock,))  # 记得把锁当作参数传递给func参数
    t.start()

RLock（递归锁）

说白了就是在一个大锁中还要再包含子锁

threading模块的Lock类，它不支持嵌套锁。RLcok类的用法和Lock一模一样，但它支持嵌套，因此我们一般直接使用RLcok类。

import threading, time

def run1():
    print("grab the first part data")
    lock.acquire()
    global num
    num += 1
    lock.release()
    return num
def run2():
    print("grab the second part data")
    lock.acquire()
    global num2
    num2 += 1
    lock.release()
    return num2
def run3():
    lock.acquire()
    res = run1()
    print(‘--------between run1 and run2-----‘)
    res2 = run2()
    lock.release()
    print(res, res2)
if __name__ == ‘__main__‘:
    num, num2 = 0, 0
    lock = threading.RLock()
    for i in range(10):
        t = threading.Thread(target=run3)
        t.start()
while threading.active_count() != 1:
    print(threading.active_count())
else:
    print(‘----all threads done---‘)
    print(num, num2)

时器（Timer）

定时器，指定n秒后执行某操作。很简单但很使用的东西。

from threading import Timer

def hello():

    print("hello, world")

t = Timer(1, hello)  # 表示1秒后执行hello函数

t.start()

信号量(Semaphore)

这种锁允许一定数量的线程同时更改数据，它不是互斥锁。比如地铁安检，排队人很多，工作人员只允许一定数量的人进入安检区，其它的人继续排队。

互斥锁 同时只允许一个线程更改数据，而Semaphore是同时允许一定数量的线程更改数据 ，比如厕所有3个坑，那最多只允许3个人上厕所，后面的人只能等里面有人出来了才能再进去。

import time
import threading
 
def run(n):
    semaphore.acquire()
    print("run the thread: %s" % n)
    time.sleep(1)
    semaphore.release()
 
num = 0
semaphore = threading.BoundedSemaphore(5)  # 最多允许5个线程同时运行
for i in range(20):
    t = threading.Thread(target=run, args=(i,))
    t.start()

事件(Event)

事件主要提供了三个方法 set、wait、clear。

事件机制：全局定义了一个“Flag”，如果“Flag”的值为False，那么当程序执行wait方法时就会阻塞，如果“Flag”值为True，那么wait方法时便不再阻塞。这种锁，类似交通红绿灯（默认是红灯），它属于在红灯的时候一次性阻挡所有线程，在绿灯的时候，一次性放行所有的排队中的线程。

clear：将“Flag”设置为False

set：将“Flag”设置为True

!/usr/bin/env python
#-*- coding:utf-8 -*-
# Author:DCC

import time
import threading
import random

event = threading.Event()

def lighter():
    count = 0
    event.set() #先设置成绿灯
    while True:
        if count > 5 and count < 10:
            #改成红灯
            event.clear() #把标志位清除
            print("\033[41;1m red.....\033[0m")
        elif count > 10:
            event.set() #设置成路灯
            count = 0
        else:
            print("\033[42;1m green \033[0m")
        time.sleep(1)
        count +=1
def car(name):
    while True:
        if event.is_set(): #代表绿灯
            print("[%s] is running " % name)
            time.sleep(2)
        else:
            print("[%s] is waitting....... " % name)
            event.wait()
            print("[%s] green light is on ,start going" % name)

light = threading.Thread(target=lighter,)
light.start()
car1 = threading.Thread(target=car,args=("Tesla",))
car1.start()

队列

通常而言，队列是一种先进先出的数据结构，与之对应的是堆栈这种后进先出的结构。但是在python中，它内置了一个queue模块，它不但提供普通的队列，还提供一些特殊的队列

Queue：先进先出队列

import queue
q = queue.Queue(5)
q.put(11)
q.put(22)
q.put(33)
 
print(q.get())
print(q.get())
print(q.get())

Queue类的参数和方法：

qsize() 获取当前队列中元素的个数，也就是队列的大小

empty() 判断当前队列是否为空，返回True或者False

full() 判断当前队列是否已满，返回True或者False

put(self, block=True, timeout=None)

get(self, block=True, timeout=None)
LifoQueue：后进先出队列

import queue
q = queue.LifoQueue()
q.put(123)
q.put(456)
print(q.get())

PriorityQueue：优先级队列

q = queue.PriorityQueue()
q.put((1,"alex1"))
q.put((1,"alex2"))
q.put((1,"alex3"))
q.put((3,"alex3"))
print(q.get())

生产者消费者模型

import time,random

import queue,threading

q = queue.Queue()

def Producer(name):

  count = 0

  while count <20:

    time.sleep(random.randrange(3))

    q.put(count)

    print(‘Producer %s has produced %s baozi..‘ %(name, count))

    count +=1

def Consumer(name):

  count = 0

  while count <20:

    time.sleep(random.randrange(4))

    if not q.empty():

        data = q.get()

        print(data)

        print(‘\033[32;1mConsumer %s has eat %s baozi...\033[0m‘ %(name, data))

    else:

        print("-----no baozi anymore----")

    count +=1

p1 = threading.Thread(target=Producer, args=(‘A‘,))

c1 = threading.Thread(target=Consumer, args=(‘B‘,))

p1.start()

c1.start()

本文出自 “东风十里柔情” 博客，请务必保留此出处http://xiaofengcanyue.blog.51cto.com/6671161/1855630