线程

一. 线程概念

1. 什么是线程

　　线程被称作轻量级的进程, 计算机的执行单位以线程为单位, 计算机的最小可执行是线程; 进程是资源分配的基本单位, 线程是可执行的基本单位，是可被调度的基本单位。

2. 线程的特点 :

  1)  GIL：全局解释锁（只有Cpython解释器才有）对于线程来说，因为有了GIL, 所以没有真正的并行

  2)  线程不可以自己独立拥有资源。线程的执行, 必须依赖于所属进程中的资源, 进程中必须至少应该有一个线程.

 3)  线程又分为用户级线程和内核级线程 :

　　 用户级线程：对于程序员来说的，这样的线程完全被程序员控制执行, 调度

　　 内核级线程：对于计算机内核来说的，这样的线程完全被内核控制调度.

3 . 线程和进程的区别

   1) 组成

　　进程由 代码段 数据段 PCB组成（process control block）

　　线程由 代码段 数据段 TCB组成（thread control block）

  2)  cpu切换进程要比cpu切换线程 慢很多, 在python中, 如果IO操作过多的话，使用多线程好. 计算密集的情况下，使用多进程好. 

   3)  在同一个进程内，所有线程共享这个进程的pid, 也就是说所有线程共享所属进程的所有资源和内存地址

  4) 在同一个进程内，所有线程共享该进程中的全局变量

  5) 因为有GIL锁的存在，在Cpython中, 没有真正的线程并行, 但是有真正的多进程并行

  6) 关于守护线程和守护进程的事情（注意：代码执行结束并不代表程序结束）

　　守护进程：要么自己正常结束, 要么根据父进程的代码执行结束而结束 .

　　守护线程：要么自己正常结束, 要么根据父线程的执行结束而结束 .

二. 线程的使用

1. 线程的创建

　　线程的创建需要导入threading模块, 具体操作和进程的创建类似.

from threading import Thread
import time
#>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>直接开启线程
def func():
    print('这是一个子线程')
    time.sleep(2)


if __name__ == '__main__':
    t = Thread(target=func,args=())
    t.start()

#>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>通过继承开启线程
class MyThread(Thread):
    def __init__(self):
        super(MyThread, self).__init__()
    def run(self):
        print('我是一个子线程')

t = MyThread()
t.start()

2. 多线程和多线程

  1) pid

　　在同一个进程里, 线程的pid都一样,和父进程的pid相同.

from multiprocessing import Process
from threading import Thread
import os


def func(name):
    print('我是一个%s，我的pid是%s'%(name,os.getpid()))


if __name__ == '__main__':

    print('我是main，我的pid是%s'%(os.getpid()))
    for i in range(10):
        p = Process(target=func,args=('进程',))
        p.start()

    for i in range(10):
        p = Thread(target=func,args=('线程',))
        p.start()

  2) 开启效率

　　线程的开启效率比进程的开启效率要高

rom multiprocessing import Process
from threading import Thread
import time



def func():
    pass

if __name__ == '__main__':
    start = time.time()
    for i in range(1000):
        p = Process(target=func)
        p.start()
    print('开100个进程的时间:', time.time() - start)

    start = time.time()
    for i in range(1000):
        p = Thread(target=func)
        p.start()
    print('开100个线程的时间:', time.time() - start)

# 开100个进程的时间: 21.032203197479248
# 开100个线程的时间: 0.17511582374572754

  3) 共享内存

　　同一进程内线程间是可以共享进程内的数据.

from threading import Thread,Lock
import time


def func():
    global num
    tmp = num
    num = tmp - 1


if __name__ == '__main__':
    num = 100
    t_l = []
    for i in range(100):
        t = Thread(target=func)
        t.start()
        t_l.append(t)
    time.sleep(1)
    [t.join() for t in t_l]
    print(num)  # 0

3. 守护线程

　　守护线程是根据主线程执行完毕而结束, 而不是根据主线程的代码执行完毕而结束.

from threading import Thread
import time

def func():
    time.sleep(2)
    print(123)

def func1():
    time.sleep(1)
    print('abc')

if __name__ == '__main__':
    t = Thread(target=func)
    t.daemon = True  # 把线程t设为守护线程
    t.start()
    t1 = Thread(target=func1)
    t1.start()
    print(99999999999999999999)

4. 线程的其它方法

Thread实例对象的方法
  # isAlive(): 返回线程是否活动的。
  # getName(): 返回线程名。
  # setName(): 设置线程名。

threading模块提供的一些方法：
  # threading.currentThread(): 返回当前的线程变量。
  # threading.enumerate(): 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前，不包括启动前和终止后的线程。
  # threading.activeCount(): 返回正在运行的线程数量，与len(threading.enumerate())有相同的结果。

三. 线程的其他操作

　　线程的操作都和进程操作一样

1. 锁 

  1) 全局解释器锁 : GIL

　　Python代码的执行由Python虚拟机(也叫解释器主循环)来控制。Python在设计之初就考虑到要在主循环中，同时只有一个线程在执行。虽然 Python 解释器中可以“运行”多个线程，但在任意时刻只有一个线程在解释器中运行。对Python虚拟机的访问由全局解释器锁(GIL)来控制，正是这个锁能保证同一时刻只有一个线程在运行。

　　在多线程环境中，Python 虚拟机按以下方式执行：

　　　　a、设置 GIL；

　　　　b、切换到一个线程去运行；

　　　　c、运行指定数量的字节码指令或者线程主动让出控制(可以调用 time.sleep(0))；

　　　　d、把线程设置为睡眠状态；

　　　　e、解锁 GIL；

　　　　d、再次重复以上所有步骤。

  2) 互斥锁 : Lock  一把锁只能配一把钥匙

　　死锁 : 

from threading import Thread,Lock
def man(l_tic, l_sea):
    l_tic.acquire()  # 拿到了票的钥匙
    print('man, ticket')
    time.sleep(0.2)
    l_sea.acquire()  # 拿不到座位的钥匙 阻塞
    print('man, seat')
    l_sea.release()
    l_tic.release()
def woman(l_tic, l_sea):
    l_sea.acquire()  # 拿到了座位的钥匙
    print('woman, seat')
    l_tic.acquire()  # 拿不到票的钥匙 阻塞
    print('woman, ticket')
    l_tic.release()
    l_sea.release()

if __name__ == '__main__':
    l_tic = Lock()
    l_sea = Lock()
    mp = Thread(target=man, args=(l_tic, l_sea))
    wp = Thread(target=woman, args=(l_tic, l_sea))
    mp.start()
    wp.start()

  3) 递归锁 : RLock  所有的锁都有一个共同的钥匙

　　解决死锁

from threading import RLock, Thread
import time
def man(l_tic, l_sea):
    l_tic.acquire()
    print('man, ticket')
    time.sleep(0.2)
    l_sea.acquire()
    print('man, ticket')
    l_sea.release()
    l_tic.release()
def woman(l_tic, l_sea):
    l_sea.acquire()
    print('woman, seat')
    l_tic.acquire()
    print('woman, ticket')
    l_tic.release()
    l_sea.release()

if __name__ == '__main__':
    l_tic = l_sea = RLock()
    # 递归锁:所有的锁都有一个共同的钥匙(公共钥匙)
    mp = Thread(target=man, args=(l_tic, l_sea))
    wp = Thread(target=woman, args=(l_tic, l_sea))
    mp.start()
    wp.start()

2. 信号量 : Semaphore(int)   多把钥匙

　　和进程一样

3. 事件 : Event

　　和进程一样

wait（）  判断is_set的bool值，如果为True，wait是非阻塞。
set      将is_set的bool值设置为True
is_set   标识
clear    将is_set的bool值设置为False

4. 条件 : Condition

　　使得线程等待，只有满足某条件时，才释放n个线程

# Condition的4个方法
# acquire()
# release()
# wait()       是指让线程阻塞住
# notify(int)  是指给wait发一个信号，让wait变成不阻塞
#              int是指，你要给多少给wait发信号

from threading import Thread, Condition
import time
def func(con, i):
    con.acquire()
    con.wait()  # 线程执行到这里，会阻塞住，等待notify发送信号，来唤醒此线程
    print('第%s个线程执行了' % i)
    con.release()

    
if __name__ == '__main__':
    con = Condition()
    for i in range(10):
        t = Thread(target=func, args=(con, i))
        t.start()
    while 1:
        con.acquire()
        num = input('>>>>')
        con.notify(int(num))  # 发送一个信号给num个正在阻塞在wait的线程，让这些线程正常执行
        con.release()
        time.sleep(1)

5. 定时器 : Timer

　　定时器,指定n秒后执行某个操作.

from threading import Timer

def func():
    print(12345)

Timer(5, func).start()  # 倒计时5秒执行func

6. 线程队列

　　queue队列 ：使用import queue，用法与进程Queue一样

from multiprocessing import Queue# 是用于多进程的队列，就是专门用来做进程间通信（IPC）。
import queue# 是用于同一进程内的队列，不能做多进程之间的通信
q = queue.Queue(num)
       num ： # 队列的最大长度
       q.get()# 阻塞等待获取数据，如果有数据直接获取，如果没有数据，阻塞等待
       q.put()# 阻塞，如果可以继续往队列中放数据，就直接放，不能放就阻塞等待 

优先级队列，put()方法接收的是一个元组（），第一个位置是优先级，第二个位置是数据
# 优先级如果是数字，直接比较数值
# 如果是字符串，是按照 ASCII 码比较的。当ASCII码相同时，会按照先进先出的原则

queue.Queue()          #先进先出
queue.LifoQueue()      #后进先出
queue.PriorityQueue()  #优先级的队列

7. 线程池:

在一个池子里，放固定数量的线程，这些线程等待任务，一旦有任务来，就有线程自发的去执行任务。

from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor   # 线程池
from multiprocessing import Pool  # 进程池  
#  concurrent.futures 这个模块是异步调用的机制
#  concurrent.futures 提交任务都是用submit
#  for + submit 多个任务的提交
#  shutdown 是等效于Pool中的close+join，是指不允许再继续向池中增加任务，然后让父进程(线程)等待池中所有进程执行完所有任务。

   1) map

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor  # 都是异步调用机制
import os
def func(num):
    num += 1
    return num


t = ThreadPoolExecutor(os.cpu_count() * 5)
ret = t.map(func, [i for i in range(200)])  # 返回值是生成器
t.shutdown()  # close + join
print(ret.__next__())

  2) submit

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor  # 都是异步调用机制
import os,time
# 多进程

def func(num):
    num += 1
    # print(num)


if __name__ == '__main__':
    tp = ProcessPoolExecutor(os.cpu_count() + 1)
    start = time.time()
    for i in range(20000):
        tp.submit(func, i)
    tp.shutdown()
    print('多进程', time.time() - start)

# 多线程

def func(num):
    num += 1
    return num


t = ThreadPoolExecutor(os.cpu_count() * 5)
lst = []
start = time.time()
for i in range(200):
    ret = t.submit(func, i)
    lst.append(ret)
t.shutdown()
print('多线程', time.time() - start)
[print(i.result()) for i in lst]

3) 线程的回调函数

# 线程的回调函数:
# t.submit(func,i).add_done_callback(函数名)
# 线程的id: current_thresd()

from concurrent.futures import ProcessPoolExecutor
# 不管是ProcessPoolExecutor的进程池  还是Pool的进程池，回调函数都是父进程调用的。
import os
import requests


def func(num):
    sum = 0
    for i in range(num):
        sum += i ** 2
    return sum

def call_back_fun(res):
    # print(res.result(),os.getpid())
    print(os.getpid())

if __name__ == '__main__':
    print(os.getpid())
    t = ProcessPoolExecutor(20)
    for i in range(1000):
        t.submit(func,i).add_done_callback(call_back_fun)
    t.shutdown()