线程

多进程多线程的join和队列的join是不一样的 进程队列 q=multiprocess.Queue() 多个进程之间能通信（不建议，一般用管道实现进程间通信） 线程队列 q=queue.Queue() 多个线程之间能通信 多进程同步中的Lock ，需要传给Process 多线程的

进程线程的特点

1. 每个进程都有自己 独立的 内存地址

2. 线程之间的切换 相对于 进程之间切换 更为方便，代价也更低。

3. 对于IO密集型的任务，使用多线程还是能提高一下CPU使用率。

4. 对于CPU密集型的任务，Python中的多线程其实是个鸡肋……没卵用……在Python的解释器CPython中存在一个互斥锁。简单来讲就是同一时间只能有一个线程在执行，其它线程都处于block模式。

5. 要想在py中充分利用多核cpu，就只能用多进程了。虽然代价高了些，但是比起并行计算带来的性能提升这些也微不足道了。

• 进程与线程的关系区别 一个程序至少有一个进程,一个进程至少有一个线程.(进程可以理解成线程的容器)进程在执行过程中拥有独立的内存单元，而多个线程共享内存，从而极大地提高了程序的运行效率。线程在执行过程中与进程还是有区别的。每个独立的线程有一个程序运行的入口、顺序执行序列和程序的出口。但是线程不能够独立执行，必须依存在应用程序中，由应用程序提供多个线程执行控制。 进程是具有一定独立功能的程序关于某个数据集合上的一次运行活动,进程是系统进行资源分配和调度的一个独立单位. 线程是进程的一个实体,是CPU调度和分派的基本单位,它是比进程更小的能独立运行的基本单位.线程自己基本上不拥有系统资源,只拥有一点在运行中必不可少的资源(如程序计数器,一组寄存器和栈)但是它可与同属一个进程的其他的线程共享进程所拥有的全部资源. 一个线程可以创建和撤销另一个线程;同一个进程中的多个线程之间可以并发执行. 进程：本质上就是一段程序的运行过程 线程：最小的执行单元（实例） 进程：最小的资源单位 程序计数器: 1.寄存器：cpu与内存中间的一级二级三级缓存，缓存存的是进程线程之间的切换时要保存状态的变量等关键数据、2.栈 一个线程可以创建和撤销另一个线程；同一个进程中的多个线程之间可以并发执行

线程

什么是线程 进程线程其实都是虚拟单位,都是用来帮助我们形象的描述某种事物

进程:资源单位
线程:执行单位
    将内存比如成工厂
    那么进程就相当于是工厂里面的车间
    而你的线程就相当于是车间里面的流水线
ps:每个进程都自带一个线程,线程才是真正的执行单位,进程只是在线程运行过程中
提供代码运行所需要的资源

为什么要有线程 开进程 1.申请内存空间 耗资源 2."拷贝代码" 耗资源

开线程
    一个进程内可以起多个线程,并且线程与线程之间数据是共享的
    ps:开启线程的开销要远远小于开启进程的开销

线程的定义 线程的出现是为了降低上下文切换的消耗，提高系统的并发性，并突破一个进程只能干一样事的缺陷，使到进程内并发成为可能。

线程也叫轻量级进程，它是一个基本的CPU执行单元，也是程序执行过程中的最小单元，由线程ID、程序计数器、寄存器集合和堆栈共同组成。线程的引入减小了程序并发执行时的开销，提高了操作系统的并发性能。线程没有自己的系统资源。

开启线程 方式1：

from threading import Thread
import time
​
def task(name):
    print('%s is running'%name)
    time.sleep(3)
    print('%s is over'%name)
# 开线程不需要在__main__代码块内 但是习惯性的还是写在__main__代码块内
t = Thread(target=task,args=('egon',))
t.start()  # 告诉操作系统开辟一个线程  线程的开销远远小于进程
# 小的代码执行完 线程就已经开启了
print('主')

方式2：继承

import threading
import time
class MyThread(threading.Thread):    
    def __init__(self, num):        
        threading.Thread.__init__(self)        
        self.num = num    
    def run(self):  # 定义每个线程要运行的函数
    #这个方法名字不能改
        print("running on number:%s" % self.num)        
        time.sleep(3)
if __name__ == '__main__':    
    t1 = MyThread(1)    
    t2 = MyThread(2)    
    t1.start()    
    t2.start()    
    print("ending......")

线程方法 join()：在子线程完成运行之前，这个子线程的父线程将一直被阻塞。 setDaemon(True)：将线程声明为守护线程，必须在start() 方法调用之前设置，我们需要的是只要主线程完成了，不管子线程是否完成，都要和主线程一起退出，这时就可以 用setDaemon方法

Thread实例的方法
# run():  线程被cpu调度后自动执行线程对象的run方法
# start():启动线程活动。
# isAlive(): 返回线程是否活动的。
# getName(): 返回线程名。
# setName(): 设置线程名。
​
threading模块提供的一些方法：
# threading.currentThread(): 返回当前的线程变量。
# threading.enumerate(): 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前，不包括启动前和终止后的线程。
# threading.activeCount(): 返回正在运行的线程数量，与len(threading.enumerate())有相同的结果。

from threading import Thread,current_thread,active_count
import time
import os
​
def task(name,i):
    print('%s is running'%name)
    # print('子current_thread:',current_thread().name)
    # print('子',os.getpid())
    time.sleep(i)
​
    print('%s is over'%name)
# 开线程不需要在__main__代码块内 但是习惯性的还是写在__main__代码块内
t = Thread(target=task,args=('egon',1))
t1 = Thread(target=task,args=('jason',2))
t.start()  # 告诉操作系统开辟一个线程  线程的开销远远小于进程
t1.start()  # 告诉操作系统开辟一个线程  线程的开销远远小于进程
t1.join()  # 主线程等待子线程运行完毕
print('当前正在活跃的线程数',active_count())
# 小的代码执行完 线程就已经开启了
print('主')
# print('主current_thread:',current_thread().name)
# print('主',os.getpid())

主线程运行结束之后需要等待子线程结束才能结束呢? """ 主线程的结束也就意味着进程的结束 主线程必须等待其他非守护线程的结束才能结束 (意味子线程在运行的时候需要使用进程中的资源,而主线程一旦结束了资源也就销毁了) """ 守护线程

from threading import Thread,current_thread
import time
​
def task(i):
    print(current_thread().name)
    time.sleep(i)
    print('GG')
# for i in range(3):
#     t = Thread(target=task,args=(i,))
#     t.start()
t = Thread(target=task,args=(1,))
t.daemon = True
t.start()
print('主')

线程间通信 1.定义变量 2.函数中，对变量进行改动时，在变量前加global(全局操作变量)

线程的互斥锁

from threading import Thread,Lock
import time
​
n = 100
​
def task(mutex):
    global  n
    mutex.acquire()
    tmp = n
    time.sleep(0.1)
    n = tmp - 1
    mutex.release()
​
t_list = []
mutex = Lock()
for i in range(100):
    t = Thread(target=task,args=(mutex,))
    t.start()
    t_list.append(t)
for t in t_list:
    t.join()
print(n)

 

并发并行、同步异步、阻塞非阻塞

• 并发 & 并行

并发：指系统具有处理多个任务（动作）的能力 并行：指系统具有 同一时刻 处理多个任务（动作）的能力，是 并发 的一种

• 同步&异步

同步：当进程执行到一个IO操作时候（等待外部数据）--等待不能干别的：同步 异步：当进程执行到一个IO操作时候（等待外部数据）--不等待可干别的：可干别的直到数据接受成功，再回来处理

• 阻塞非阻塞

阻塞: 阻塞调用是指调用结果返回之前，当前线程会被挂起（线程进入非可执行状态，在这个状态下，cpu不会给线程分配时间片，即线程暂停运行）。函数只有在得到结果之后才会返回。 非阻塞：非阻塞和阻塞的概念相对应，指在不能立刻得到结果之前，该函数不会阻塞当前线程，而会立刻返回。

CPU计算任务切换条件：遇到IO堵塞 或 时间轮训

问题：多核没用上 GIL：全局解释锁 在cpython解释器上 同一时刻，只有一个线程被CPU执行

任务：IO密集型 计算密集型 对于IO密集型任务，Python的多线程是有意义的；对计算密集型的任务，Python的多线程是没意义的 解决办法：多进程+协程（进程消耗大，通信麻烦，却是利用多核的唯一方法） 对于计算密集型任务，python就不适用了

1. 同步锁 等的时间太长了，每一个线程拿到的值都是初始值，导致真实的函数只执行了一次 多个线程都在同时操作同一个共享资源，所以造成了资源破坏，怎么办呢？(join会造成串行，失去多线程的意义)我们可以通过同步锁来解决这种问题

import threading
import time
def sub():
    global num
   # num-=1
    temp=num
    time.sleep(1)
    num=temp-1
   
num=100
l=[]
for i in range(100):
    t=threading.Thread(target=sub)
    t.start()
    l.append(t)
for t in l:
    t.join()
print(num)
​
##################
答案会因 time.sleep(1) 时间长短的不同而不同

解决后

import threading
import time
def sub():
    global num
   # num-=1
    print ("ok")
    lock.acquire()
    temp=num
    time.sleep(1)
    num=temp-1
    lock.release()
​
num=100
​
l=[]
lock=threading.Lock()
​
for i in range(100):
    t=threading.Thread(target=sub)
    t.start()
    l.append(t)
​
for t in l:
    t.join()
​
print(num)

2. 死锁 递归锁 死锁 在线程间共享多个资源的时候，如果两个线程分别占有一部分资源并且同时等待对方的资源，就会造成死锁，因为系统判断这部分资源都正在使用，所有这两个线程在无外力作用下将一直等待下去。下面是一个死锁的例子：

import  threading
import time
​
class MyThread(threading.Thread):
​
    def actionA(self):
        A.acquire() 
        print(self.name,"gotA",time.ctime())
        time.sleep(2)
        
        B.acquire() 
        print(self.name, "gotB", time.ctime())
        time.sleep(1)
​
        B.release() 
        A.release() 
​
    def actionB(self):
​
        B.acquire()
        print(self.name, "gotB", time.ctime())
        time.sleep(2)
​
        A.acquire()
        print(self.name, "gotA", time.ctime())
        time.sleep(1)
​
        A.release()
        B.release()
​
    def run(self):
        self.actionA()
        self.actionB()
if __name__ == '__main__':
​
    A=threading.Lock()
    B=threading.Lock()
​
    #r_lcok=threading.RLock()
    L=[]
    
    for i in range(5):
        t=MyThread()
        t.start()
        L.append(t)
        
    for i in L:
        i.join()
    print("ending....")

递归锁 为了支持在同一线程中多次请求同一资源，python提供了“可重入锁”：threading.RLock。RLock内部维护着一个Lock和一个counter变量，counter记录了acquire的次数，从而使得资源可以被多次acquire。直到一个线程所有的acquire都被release，其他的线程才能获得资源。

import  threading
import time
class MyThread(threading.Thread):
​
    def actionA(self):
        r_lcok.acquire() #count=1
        print(self.name,"gotA",time.ctime())
        time.sleep(2)
        
        r_lcok.acquire() #count=2
        print(self.name, "gotB", time.ctime())
        time.sleep(1)
        
        r_lcok.release() #count=1
        r_lcok.release() #count=0
    def actionB(self):
        r_lcok.acquire()
        print(self.name, "gotB", time.ctime())
        time.sleep(2)
​
        r_lcok.acquire()
        print(self.name, "gotA", time.ctime())
        time.sleep(1)
        
        r_lcok.release()
        r_lcok.release()
    def run(self):
        self.actionA()
        self.actionB()
​
if __name__ == '__main__':
    # A=threading.Lock()
    # B=threading.Lock()
    r_lcok=threading.RLock()
    L=[]
    
    for i in range(5):
        t=MyThread()
        t.start()
        L.append(t)
        
    for i in L:
        i.join()
    print("ending....")

--->>>

Thread-1 gotA Tue Jun 18 23:43:40 2019
Thread-1 gotB Tue Jun 18 23:43:42 2019
Thread-1 gotB Tue Jun 18 23:43:43 2019
Thread-1 gotA Tue Jun 18 23:43:45 2019
Thread-3 gotA Tue Jun 18 23:43:46 2019
Thread-3 gotB Tue Jun 18 23:43:48 2019
Thread-3 gotB Tue Jun 18 23:43:49 2019
Thread-3 gotA Tue Jun 18 23:43:51 2019
Thread-5 gotA Tue Jun 18 23:43:52 2019
Thread-5 gotB Tue Jun 18 23:43:54 2019
Thread-2 gotA Tue Jun 18 23:43:55 2019
Thread-2 gotB Tue Jun 18 23:43:57 2019
Thread-2 gotB Tue Jun 18 23:43:58 2019
Thread-2 gotA Tue Jun 18 23:44:00 2019
Thread-5 gotB Tue Jun 18 23:44:01 2019
Thread-5 gotA Tue Jun 18 23:44:03 2019
Thread-4 gotA Tue Jun 18 23:44:04 2019
Thread-4 gotB Tue Jun 18 23:44:06 2019
Thread-4 gotB Tue Jun 18 23:44:07 2019
Thread-4 gotA Tue Jun 18 23:44:09 2019
ending....

相当于acquire()就+1，acquire()就+1，release()就-1 ，release()就-1，只要>0，其他外部进程就进不去

import time
import threading
​
class Account:
    def __init__(self, _id, balance):
        self.id = _id
        self.balance = balance
        self.lock = threading.RLock()
​
    def withdraw(self, amount):
        with self.lock:
            self.balance -= amount
​
    def deposit(self, amount):
        with self.lock:
            self.balance += amount
​
    def drawcash(self, amount):#lock.acquire中嵌套lock.acquire的场景
        with self.lock:
            interest=0.05
            count=amount+amount*interest
            self.withdraw(count)
​
def transfer(_from, to, amount):
    #锁不可以加在这里 因为其他的其它线程执行的其它方法在不加锁的情况下数据同样是不安全的
     _from.withdraw(amount)
     to.deposit(amount)
​
alex = Account('alex',1000)
yuan = Account('yuan',1000)
​
t1=threading.Thread(target = transfer, args = (alex,yuan, 100))
t1.start()
​
t2=threading.Thread(target = transfer, args = (yuan,alex, 200))
t2.start()
​
t1.join()
t2.join()
​
print('>>>',alex.balance)
print('>>>',yuan.balance)

3. 信号量和同步对象

同步对象 event.set()：设置一个事件 event.clear(): 清除一个事件 event.wait()：当没有事件时，wait等待事件，当设置了一个事件，变成pass，进行下一步操作

import threading,time
class Boss(threading.Thread):
​
    def run(self):
        print("BOSS：今晚大家都要加班到22:00。")
        print(event.isSet())# False
        event.set()
        time.sleep(5)
        print("BOSS：<22:00>可以下班了。")
        print(event.isSet())
        event.set()
​
class Worker(threading.Thread):
    def run(self):
        event.wait()#    等待，直到event被设定，才继续
​
        print("Worker：哎……命苦啊！")
        time.sleep(1)
        event.clear()   #清除已设定的event
        event.wait()    #等待，直到下一个event被设定，才继续
        print("Worker：OhYeah!")
​
if __name__=="__main__":
    event=threading.Event()
​
    threads=[]
    for i in range(5):
        threads.append(Worker())
    threads.append(Boss())
    for t in threads:
        t.start()
    for t in threads:
        t.join()
​
    print("ending.....")

信号量 控制线程并发数 信号量用来控制线程并发数的，BoundedSemaphore或Semaphore管理一个内置的计数 器，每当调用acquire()时-1，调用release()时+1。      计数器不能小于0，当计数器为 0时，acquire()将阻塞线程至同步锁定状态，直到其他线程调用release()。(类似于停车位的概念)      BoundedSemaphore与Semaphore的唯一区别在于前者将在调用release()时检查计数 器的值是否超过了计数器的初始值，如果超过了将抛出一个异常。

import threading,time
class myThread(threading.Thread):
    def run(self):
        if semaphore.acquire():
            print(self.name)
            time.sleep(5)
            semaphore.release()
if __name__=="__main__":
    semaphore=threading.Semaphore(5)#同时最大5个线程并发（默认为1）
    thrs=[]
    for i in range(100):
        thrs.append(myThread())
    for t in thrs:
        t.start()

4. 队列--生产者消费者模型 多线程利器--------队列 存在的意义：列表是不安全的数据结构

• 先进先出

import queue  #  线程队列
q=queue.Queue(3)  # FIFO模式
​
q.put(12)
q.put("hello")
q.put({"name":"yuan"})
q.put_nowait(56)#  q.put(block=False)
​
print(q.qsize())  #剩余的元素
print(q.empty())  #是否为空
print(q.full())  #是否满了
# q.put(34,False)
while 1:
    data=q.get()
    print(data)
    print("----------")

• 先进后出

import queue
q=queue.LifoQueue()
​
q.put(12)
q.put("hello")
q.put({"name":"yuan"})
​
while 1:
    data=q.get()
    print(data)
    print("----------")

• 优先级 数字小的先出

q=queue.PriorityQueue()
q.put([3,12])
q.put([2,"hello"])
q.put([4,{"name":"yuan"}])
​
while 1:
    data=q.get()
    print(data[1])
    print("----------")

此包中的常用方法(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

import threading,queue,time,random
​
#生产者
def product(id, q):
    while True:
        num = random.randint(0, 10000)
        q.put(num)
        print("生产者%d生产了%d数据放入了队列" % (id, num))
        time.sleep(3)
    #任务完成
    q.task_done()
​
#消费者
def customer(id, q):
    while True:
        item = q.get()
        if item is None:
            break
        print("消费者%d消费了%d数据" % (id, item))
        time.sleep(2)
    # 任务完成
    q.task_done()
​
if __name__ == "__main__":
    # 消息队列
    q = queue.Queue()
    # 启动生产者
    for i in range(4):
        threading.Thread(target=product, args=(i,q)).start()
    # 启动消费者
    for i in range(3):
        threading.Thread(target=customer, args=(i,q)).start()

例2

import time,random
import queue,threading
​
q = queue.Queue()
​
def Producer(name):
  count = 0
  while count <10:
    print("making........")
    time.sleep(5)
    q.put(count)
    print('Producer %s has produced %s baozi..' %(name, count))
    count +=1
    q.task_done() #告诉队列，发完了
    # q.join()
    print("ok......")
​
def Consumer(name):
    count = 0
    while count <10:
        time.sleep(random.randrange(4))
    # if not q.empty():
        print("waiting.....")
#  q.join()阻塞等待队列中任务全部处理完毕，需要配合queue.task_done使用        
        q.join()
        data = q.get()
        print("eating....")
        time.sleep(4)
​
        # q.task_done()
        #print(data)
        print('33[32;1mConsumer %s has eat %s baozi...33[0m' %(name, data))
    # else:
    #     print("-----no baozi anymore----")
        count +=1
​
p1 = threading.Thread(target=Producer, args=('A君',))
c1 = threading.Thread(target=Consumer, args=('B君',))
c2 = threading.Thread(target=Consumer, args=('C君',))
c3 = threading.Thread(target=Consumer, args=('D君',))
​
p1.start()
c1.start()
c2.start()
c3.start()

如果线程里每从队列里取一次，但没有执行task_done()，则join无法判断队列到底有没有结束，在最后执行个join()是等不到结果的，会一直挂起。 可以理解为，每task_done一次 就从队列里删掉一个元素，这样在最后join的时候根据队列长度是否为零来判断队列是否结束，从而执行主线程。

让q.join()函数能判断出队列还剩多少，是否清空了

注意： put队列完成的时候千万不能用task_done() 否则会报错：task_done() called too many times 因为该方法仅仅表示get成功后，执行的一个标记。

from multiprocessing import Queue, Process
import time
​
def producer(q):
    for i in range(10):
​
        print("生产了%d号包子" % i)
        q.put(i)
        time.sleep(0.1)
​
def consumer(q):
    while True:
        # time.sleep(1)
        s = q.get()
        if s == None:break
        print("吃了%d号包子" % s)
​
if __name__ == '__main__':
    q = Queue()
    p = Process(target=producer, args=(q,))
​
    c1 = Process(target=consumer, args=(q,))
​
    p.start()
    c1.start()
    
    p.join() #生产者已经生产完了
    q.put(None) #放一个空，消费者受到也跳出循环
    print(1)

要避免的的问题是主进程永远不结束，原因是：生产者p在生产完后就结束了，但是消费者c在取空了q之后，则一直处于死循环中且卡在q.get()这一步。解决方式无非是让生产者在生产完毕后，往队列中再发一个结束信号，这样消费者在接收到结束信号后就可以break出死循环