002---线程

线程

什么是线程

线程： 线程是操作系统调度的最小单位，它包含在进程中。

比喻：一条流水线工作的流程，一条流水线必须属于一个车间，一个车间的工作过程是一个进程，车间负责把资源整合到一起，是一个资源单位，而一个车间内至少有一个流水线，流水线的工作需要电源，电源就相当于cpu

线程与进程的区别

创建一个进程，就是创建一个车间，涉及到申请空间，而且在该空间内建至少一条流水线，但创建线程，就只是在一个车间内造一条流水线，无需申请空间，所以创建开销小。

• 同一个进程内的多个线程共享该进程内的地址资源
• 创建线程的开销要远小于创建进程的开销
• 线程的开启速度更快

开启线程的两种方式

• 普通方式

  from threading import Thread
  
  
  def func(*args, **kwargs):
      print(args, kwargs)
      print('子线程启动啦')
  
  
  t = Thread(target=func, args=(1, 2, 3), kwargs={"name": 'jiangwei'})
  t.start()
  print('开启了一个线进程')
  

• 类创建

  from threading import Thread
  
  
  class MyThread(Thread):
      def __init__(self, i):
          super(MyThread, self).__init__()
          self.i = i
  
      def run(self):
          time.sleep(1)
          print(self.i)
  
  
  t = MyThread(10)
  t.start()
  
  

线程相关方法和属性

Thread对象的属性和方法

• isAlive(): 返回线程是否活动的。
• getName(): 返回线程名。
• setName(): 设置线程名。

threading模块提供的一些方法：

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

守护线程

无论是进程还是线程，都遵循：守护xxx会等待主xxx运行完毕之后被销毁

注意：运行完毕并非终止运行

• 对进程来说：指主进程代码运行完毕
• 对线程来说：指主线程所在的进程内所有的非守护线程都运行完毕，主线程才算运行完毕，因为主线程的结束代表主进程的结束，进程整体的资源都将被回收，而进程必须保证非守护线程都运行完毕才能结束。

def f1():
    print('f1', 666)
    time.sleep(3)
    print('你都没死，我咋能先死，你先把其他人杀了再说')
    time.sleep(2.5)
    print('我发现我已经死了')


def f2():
    print('f2', 999)
    time.sleep(5)
    print('哈哈哈，他把我杀了')


t1 = Thread(target=f1)
t2 = Thread(target=f2)
t1.daemon = True  # 守护线程t1
t1.start()
t2.start()
print('主')       # 主线程结束(等待其他非守护进程结束，也就是f2运行完毕)、守护进程随之结束(f1不会执行到print)   
"""
f1 666
f2 999
主
你都没死，我咋能先死，你先把其他人杀了再说
哈哈哈，他把我杀了
"""

GIL全局解释器锁

• 本质也是一个互斥锁，不是Python的特性，是Cpython解释器的特性
• 在同一个进程下开启的多线程，同一时刻只能有一个线程能拿到那把锁，只有一个线程然后去执行
• 不同的锁保护不同的数据，自定义的互斥锁，保护程序内的数据，而GIL锁是解释器级别的，与垃圾回收的数据有关
• 开启多线程之后，先抢GIL,后抢自定义锁

为什么要有这把锁

解释器的代码是共享的，如果程序中有一个线程是修改全局变量n=100,而解释器里的垃圾回收线程执行回收n=100的操作，这就导致了数据混乱，不安全。

疑问

有了GIL的存在，同一时刻同一进程中只有一个线程被执行。那么，进程可以利用多核，但开销大，而线程开销小，却无法利用多核？

其实不然：
我开了一个工厂。假设我的原材料充足的情况下，我的工人越多越好，效率快。类似于cpu做计算.但是，如果我的原料不充足，要从西伯利亚运过来，那么我的工人越多越好吗？无济于事。类似于cpu做io读写操作。

结论

• 多线程用于IO密集型，如socket，爬虫，web
• 多进程用于计算密集型，如金融分析，科学计算

死锁与递归锁

死锁

两个或两个以上的进程或线程在执行过程中，因为争夺资源而造成的互相等待的现象。若无外力作用，他们将无法推进下去，会进入死锁状态

递归锁Rlock

可以多次acquire,内部维护一个Lock和counter变量。counter记录acquire的次数，每加锁一次就+1。直到该线程的所有锁被释放，其他线程才能抢到

import time
from threading import Lock, Thread, RLock

# l1 = Lock()
# l2 = Lock()


# 递归锁  可以连续acquire多次，每acquire一次，计数器+1  只要计数不为0，就不能被其他线程抢到
l1 = l2  = RLock()

class MyThread(Thread):
    def run(self):
        self.f1()
        self.f2()

    def f1(self):
        l1.acquire()
        print('%s拿到了l1' % self.name)
        l2.acquire()
        print('%s拿到了l2' % self.name)
        l2.release()
        l1.release()

    def f2(self):
        l2.acquire()
        print('%s拿到了l1' % self.name)
        time.sleep(0.1)
        l1.acquire()
        print('%s拿到了l2' % self.name)
        l1.release()
        l2.release()


if __name__ == '__main__':
    for i in range(10):
        t = MyThread()
        t.start()

信号量

也是一把锁，放多个钥匙。只不过不再是家里的卫生间，而是公共厕所，每次能进入多人。

import time, random
from threading import Thread, Semaphore, currentThread

sm = Semaphore(5)


def task():
    with sm:
        print('%s ing ' % currentThread().getName())
        time.sleep(random.randint(1, 3))


if __name__ == '__main__':
    for i in range(10):
        t = Thread(target=task)
        t.start()

事件

主要是根据状态来控制线程

常用方法

• e.isSet()/e.is_set()：返回event的状态值；
• e.wait()：如果 event.isSet()==False将阻塞线程；
• e.set()： 设置event的状态值为True，所有阻塞池的线程激活进入就绪状态，等待操作系统调度；
• e.clear()：恢复event的状态值为False；

import time
from threading import Thread, Event

event = Event()


def student(name):
    print('%s 正在听课' % name)
    event.wait()
    # event.wait(2)  # 时间
    print('%s 课间活动' % name)


def teacher(name):
    print('%s 正在讲课' % name)
    time.sleep(7)

    event.set()  # 改变状态为True，默认false event.clear()设为false


if __name__ == '__main__':
    t = Thread(target=teacher, args=('jw',))
    s1 = Thread(target=student, args=('alex',))
    s2 = Thread(target=student, args=('wupeiqi',))
    s3 = Thread(target=student, args=('egon',))
    t.start()
    s1.start()
    s2.start()
    s3.start()

定时器

from threading import Timer

def task(name):
    print('hello %s'%name)

t = Timer(3,task,args=('egon',))
t.start()

线程队列

import queue

q = queue.Queue() # 先进先出--队列

q1 = queue.LifoQueue()  # 后进先出--堆栈

q2 = queue.PriorityQueue()  # 优先级队列

q2.put('a')
q2.put('c')
q2.put('a')

print(q2.get())
print(q2.get())
print(q2.get())

线程池

基本方法

• t.submit(func,*args,**kwargs)：异步提交任务
• t.shutdown(wait=True)：pool.close() + pool.join()
• t.result()：拿结果
• t.add_done_callback(func)：添加回调函数

使用

import time, os, random

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor


def task(name):
    print('name: %s pid:%s run' % (name, os.getpid()))
    time.sleep(random.randint(1, 3))
    return '我是回调函数'


def call_back(m):
    print(m.result())


if __name__ == '__main__':
    pool = ThreadPoolExecutor(5)
    t_list = []
    for i in range(10):
        t = pool.submit(task, 'alex%s' % i)
        t.add_done_callback(call_back)
        t_list.append(t)
    pool.shutdown(wait=True)  # close + join
    print('主')
    for t in t_list:
        print('----', t.result())