python--线程知识详解

Threading用于提供线程相关的操作，线程是应用程序中工作的最小单元。

1.1、threading模块

threading模块建立在_thread模块之上。thread模块以低级=原始的方式来处理和控制线程，而threading模块
通过对thread进行二次封装，提供了更方便的api来处理线程。

简单的线程实例：
创建了20个“前台”线程，然后控制器就交给了CPU，CPU根据指定算法进行调度，分片执行指令

#!/usr/bin/env python
# -*- coding:utf-8 -*-
"""
创建一个简单的threading线程实例
"""
import threading
import time

def to_worker(num):
    """
    线程方法
    :param num:
    :return:
    """
    time.sleep(1)
    print("The num is %s" % num)
    return

for i in range(5):
    t = threading.Thread(target=to_worker, args=(i, ))
    t.start()  #激活线程

代码执行结果：

1.2、创建线程的构造方法

t = threading.Thread(group = None, target = None, name = Nome, args = 0, kwargs = {})

注释说明：
group --线程组
target --要执行的方法
name --线程名
args/kwargs -要传入方法的参数

Thread类提供了以下方法:
1、t.start() --激活线程
2、t.getName() --获取线程的名称
3、t.setName() --设置线程的名称
4、t.name() --获取或设置线程的名称
5、t.is_alive() --判断线程是否为激活状态
6、t.isAlive() --判断线程是否为激活状态
7、t.setDaemon() --设置为后台线程或前台线程（默认：False）
8、t.isDaemon() --判断是否为守护线程
9、t.ident() --获取线程的标识符。线程标识符是一个非零整数，只有在调用start()方法后，该属性才有效，否则它只返回None
10、t.join() --逐个执行每个线程，执行完毕后继续往下执行，该方法使得多线程变得无意义。
11、t.run() --线程被cpu调度后自动执行线程对象的run方法

1.3、python线程锁

当有一个数据有多个线程对其进行修改的时候，任何一个线程改变他都会对其他线程造成影响，如果我们想某一个线程在使用完之前，其他线程不能对其修改，就需要对这个线程加一个线程锁。

简单的线程锁实例：

#!/usr/bin/env python
# -*- coding:utf-8 -*-
"""
线程锁小实例
"""
import threading
import time

globals_num = 0

lock = threading.RLock()

def Func():
    lock.acquire()  #获取锁
    global globals_num
    globals_num += 1
    time.sleep(1)
    print(globals_num)
    lock.release()  #释放锁

for i in range(10):
    t = threading.Thread(target=Func)
    t.start()

代码执行结果：

threading.RLock和threading.Lock 的区别
RLock允许在同一线程中被多次acquire。而Lock却不允许这种情况。 如果使用RLock，那么acquire和release必须成对出现，即调用了n次acquire，必须调用n次的release才能真正释放所占用的琐。

import threading
lock = threading.Lock()    #Lock对象
lock.acquire()
lock.acquire()  #产生了死琐。
lock.release()
lock.release()　



import threading
rLock = threading.RLock()  #RLock对象
rLock.acquire()
rLock.acquire()    #在同一线程内，程序不会堵塞。
rLock.release()
rLock.release()

1.4、threading.Event

1、python线程的事件用于主线程控制其他线程的执行，事件主要提供了三个方法：set、wait、clear。
2、事件处理的机制：全局定义了一个Flag，如果Flag值为False，那么当程序执行event.wait方法时就会阻塞，
如果Flag值为True，那么event.wait方法时便不再阻塞。

方法说明：
clear --将Flag设置为False
set -- 将Flag设置为True
Event.isSet() --判断标识符是否为True


threading.Event简单实例：
当线程执行的时候，如果flag为False，则线程会阻塞，当flag为True的时候，线程不会阻塞。它提供了本地和远程的并发性。

#!/usr/bin/env python
# -*- coding:utf-8 -*-
"""
threading.Event事件实例
"""
import threading

def do(event):
    print("start.....")
    event.wait()
    print("execuse...")

event_obj = threading.Event()

for i in range(5):
    t = threading.Thread(target=do, args=(event_obj,))
    t.start()

event_obj.clear()
inp = input('input:(true) ')
if inp == "true":
    event_obj.set()

代码执行结果：

1.5、threading.Condition(条件变量)

示例说明：当小伙伴a在往火锅里面添加鱼丸，这个就是生产者行为；另外一个小伙伴b在吃掉鱼丸就是消费者行为。当火锅里面鱼丸达到一定数量加满后b才能吃，这就是一种条件判断了。

Condition（条件变量）通常与一个锁关联。需要在多个Contidion中共享一个锁时，可以传递一个Lock/RLock实例给构造方法，否则它将自己生成一个RLock实例。

可以认为，除了Lock带有的锁定池外，Condition还包含一个等待池，池中的线程处于状态图中的等待阻塞状态，直到另一个线程调用notify()/notifyAll()通知；得到通知后线程进入锁定池等待锁定。

Condition():

acquire(): 线程锁
release(): 释放锁
wait(timeout): 线程挂起，直到收到一个notify通知或者超时（可选的，浮点数，单位是秒s）才会被唤醒继续运行。wait()必须在已获得Lock前提下才能调用，否则会触发RuntimeError。
notify(n=1): 通知其他线程，那些挂起的线程接到这个通知之后会开始运行，默认是通知一个正等待该condition的线程,最多则唤醒n个等待的线程。notify()必须在已获得Lock前提下才能调用，否则会触发RuntimeError。notify()不会主动释放Lock。
notifyAll(): 如果wait状态线程比较多，notifyAll的作用就是通知所有线程。

生产者与消费者示例：
现实场景：当a同学王火锅里面添加鱼丸加满后（最多3个，加满后通知b去吃掉），通知b同学去吃掉鱼丸（吃到0的时候通知a同学继续添加）

#!/usr/bin/env python
# -*- coding:utf-8 -*-
"""
threading.Condition
"""
import threading
import time

con = threading.Condition()

num = 0

#生产者
class Producer(threading.Thread):

    def __init__(self):
        threading.Thread.__init__(self)

    def run(self):
        #锁定线程
        global  num
        con.acquire()  #获取锁
        while True:
            print("a同学开始添加......")
            num += 1
            print("锅里丸子个数为：%s" % str(num))
            time.sleep(1)
            if num >= 3:
                print("丸子个数已经达到3个了，无法添加。")
                #唤醒等待的线程
                con.notify()  #唤醒同学开吃
                #等待通知
                con.wait()

        #释放锁
        con.release()

#消费者
class Consumers(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)

    def run(self):
        con.acquire()
        global num
        while True:
            print("我准备开吃了...")
            num -= 1
            print("锅里丸子数量为:%s" % str(num))
            time.sleep(2)
            if num <= 0:
                print("丸子吃完了，赶紧添加啦..")
                con.notify()  #唤醒等待的线程
                #等待通知
                con.wait()
        con.release()  #释放锁

p = Producer()
c = Consumers()
p.start()
c.start()

代码执行结果：

1.6、Queue模块

1.6.1、创建一个“队列”对象
import queue
q = queue.queue(maxsize = 10)
queue.queue类即是一个队列的同步实现。队列长度可为无限或者有限。可通过queue的构造函数的可选参数maxsize来设定队列长度。如果maxsize小于1就表示队列长度无限。

1.6.2、将一个值放入队列中
q.put(10)
调用队列对象的put()方法在队尾插入一个项目。put()有两个参数，第一个item为必需的，为插入项目的值；第二个block为可选参数，默认为
1。如果队列当前为空且block为1，put()方法就使调用线程暂停,直到空出一个数据单元。如果block为0，put方法将引发Full异常。

1.6.3、将一个值从队列中取出
q.get()
调用队列对象的get()方法从队头删除并返回一个项目。可选参数为block，默认为True。如果队列为空且block为True，get()就使调用线程暂停，直至有项目可用。如果队列为空且block为False，队列将引发Empty异常。

1.6.4、Python queue模块有三种队列及构造函数:
1、Python queue模块的FIFO队列先进先出。 class queue.queue(maxsize)
2、LIFO类似于堆，即先进后出。 class queue.Lifoqueue(maxsize)
3、还有一种是优先级队列级别越低越先出来。 class queue.Priorityqueue(maxsize)

1.6.5、queue常用方法(q =queue.queue()):
q.qsize() 返回队列的大小
q.empty() 如果队列为空，返回True,反之False
q.full() 如果队列满了，返回True,反之False
q.full 与 maxsize 大小对应
q.get([block[, timeout]]) 获取队列，timeout等待时间
q.get_nowait() 相当q.get(False)
非阻塞 q.put(item) 写入队列，timeout等待时间
q.put_nowait(item) 相当q.put(item, False)
q.task_done() 在完成一项工作之后，q.task_done() 函数向任务已经完成的队列发送一个信号
q.join() 实际上意味着等到队列为空，再执行别的操作

1.6.6、简单的queue实例：生产者-消费者模型

#!/usr/bin/env python
# -*- coding:utf-8 -*-
"""
Queue队列
"""
import queue
import threading


message = queue.Queue(10)

def producer(i):
    while True:
        message.put(i)

def consumer(i):
    while True:
        msg = message.get(i)

for i in range(12):
    t = threading.Thread(target=producer, args=(i, ))
    t.start()
for i in range(10):
    t = threading.Thread(target=consumer, args=(i, ))
    t.start()

1.7、自定义线程池

1.7.1、方法一：简单往队列中传输线程数

#!/usr/bin/env python
# -*- coding:utf-8 -*-
"""
自定义线程池
方法一：简单往队列中传输线程数
"""
import threading
import time
import queue

class ThreadingPool():
    def __init__(self, max_num = 10):
        self.queue = queue.Queue(max_num)
        for i in range(max_num):
            self.queue.put(threading.Thread)

    def getthreading(self):
        return self.queue.get()

    def addthreading(self):
        self.queue.put(threading.Thread)


def func(p, i):
    time.sleep(1)
    print(i)
    p.addthreading()

if __name__ == "__main__":
    p = ThreadingPool()
    for i in range(12):
        thread = p.getthreading()
        t = thread(target = func, args = (p, i))
        t.start()

代码执行结果：

1.7.2、方法二：往队列中无限添加任务

#!/usr/bin/env python
# -*- coding:utf-8 -*-
"""
自定义线程池
方法二：往队列中无限添加任务
"""
import queue
import threading
import contextlib
import time

StopEvent = object()

class ThreadPool(object):

    def __init__(self, max_num):
        self.q = queue.Queue()
        self.max_num = max_num

        self.treminal = False
        self.generate_list = []
        self.free_list = []

    def run(self, func, args, callback=None):
        """
        线程池执行一个任务
        :param func: 任务函数
        :param args: 任务函数所需参数
        :param callback: 任务执行失败或成功后执行的回调函数，回调函数有两个参数：1、任务函数执行状态；2、任务函数返回值（默认为None,即不执行回调函数）
        :return:如果线程池已经终止，则返回True，否则为None
        """
        if len(self.free_list) == 0 and len(self.generate_list) < self.max_num:
            self.generate_thread()
            w = (func, args, callback,)
            self.q.put(w)

    def generate_thread(self):
        """
        创建一个线程
        :param self:
        :return:
        """
        t = threading.Thread(target=self.call)
        t.start()

    def call(self):
        """
        循环去获取任务函数并执行任务函数
        :return:
        """
        current_thread = threading.currentThread
        self.generate_list.append(current_thread)

        event = self.q.get()  #获取线程
        while event != StopEvent: #判断获取的线程数不等于全局变量
            func, arguments, callback = event  #拆分元组， 获取执行函数，参数， 回调函数
            try:
                result = func(*arguments) #执行函数
                status = True

            except Exception as e: #函数执行失败
                status = False
                result = e

            if callback is not None:
                try:
                    callback(status, result)
                except Exception as e:
                    pass

            with self.work_state():
                event = self.q.get()

        else:
            self.generate_list.remove(current_thread)


    def close(self):
        """
        关闭线程，给传输全局非元组的变量来进行关闭
        :return:
        """
        for i in range(len(self.generate_list)):
            self.q.put(StopEvent)


    def terminate(self):
        """
        突然关闭线程
        :return:
        """
        self.terminal = True
        while self.generate_list:
            self.q.put(StopEvent)
        self.q.empty()


    def work_state(self):
        self.free_list.append(threading.current_thread)
        try:
            yield
        finally:
            self.free_list.remove(threading.currentThread)

def work(i):
    print(i)
    return i + 1 #返回给回调函数

def callback(ret):
    print(ret)

pool = ThreadPool(10)
for item in range(50):
    pool.run(func=work, args=(item, ), callback=callback)

pool.terminate()