##内容回顾

# JoinableQueue

可以被join的队列

join是等待任务结束  

队列怎么叫结束?

​    调用task_done一次则表示有一个数据被处理完成了  当task_done次数等于put的次数就意味着任务处理完成了   

​    这也是join的执行时机 

该队列可以明确告知数据的使用方,所有数据都已经处理完成  

在生产者消费者模型中,解决了消费者不知道何时算是任务结束的问题  

具体过程:主进程先等待所有的生产者进程生成完毕,再等队列中的数据被全部处理,这就意味着,任务全部结束  



# 多线程

​    使用多线程 多进程的目的 是一致 ,都是为了并发执行任务,从而提高效率   

​    什么是线程:

线程是操作系统运算调度的最小单位   (CPU最小执行单位),线程被包含在进程中,一个线程就是一个固定的执行流程 (控制流)

​    线程的特点:

​        进程是不能被执行,进程是一个资源单位,其中包含了程序运行所需要的所有资源,

​        线程才是真正的执行单位,光有进程程序是无法运行的,必须先创建进程将资源加载到进程中,在启动线程来执行任务  

​        一个进程至少包含一个线程,称之为主线程,主线程是由操作系统来开启,  

​        一个进程可以包含多个线程,来提高程序的效率      

​    线程与进程的区别:

​        线程创建的开销远小于进程 

​        统一进程中的所有线程共享进程内的资源  

​        线程之间没有父子关系,都是平等的,PID相同  

如何选择:

​    要根据具体的任务类型,IO密集    计算密集   



## 线程的使用方法与进程一模一样  

开启线程的位置可以是任何位置  

# 守护线程

​    守护线程会在所有非守护线程结束时一起结束   ,当然守护可以提前结束  

# 线程安全问题

​    并发操作同一个资源 可能导致数据错乱的问题    

​    解决方案: 加互斥锁 

​     

​    互斥锁  mutex

​    递归锁   Rlock 同一线程可以多次执行acquire()

​    信号量 semaphore  

​    死锁问题:

​        不止一个锁,分别被不同线程持有,相互等待对方释放,就会导致锁死问题  

 

　　##基于多线程实现并发的套接字案例

---------------------------服务器.py----------------------------------------------
import socket
from concurrent.futures import ThreadPoolExecutor

pool = ThreadPoolExecutor(5)


def working(client):
    while True:
        try:
            data = client.recv(1024)
            if not data:
                client.close()
                break
            client.send(data.upper())
        except ConnectionResetError:
            client.close()
            break

def server():
    server = socket.socket()
    server.bind(("127.0.0.1", 1688))
    server.listen()
    while True:
        client, addr = server.accept()
        # 把任务提交到线程池
        pool.submit(working,client)
        # working(client)
server()

---------------------------客户端.py----------------------------------------------
import socket

client = socket.socket()
client.connect(("127.0.0.1",1688))


while True:
    msg = input(":").strip()
    if not msg:continue
    client.send(msg.encode("utf-8"))
    data = client.recv(1024).decode("utf-8")
    print(data)

　　##GIL 全局解释器锁

#1、什么是GIL？

“”“
在CPython中，这个全局解释器锁，也称为GIL,是一个互斥锁,防止多个线程在同一时间执行Python字节码,这个锁是非常重要的,因为CPython的内存管理非线程安全的,很多其他的特性依赖于GIL,所以即使它影响了程序效率也无法将其直接去除。

需要知道的是,解释器并不只有CPython,还有PyPy,JPython等等。GIL也仅存在与CPython中,这并不是Python这门语言的问题，而是CPython解释器的问题！

”“”

#2、GIL与GC的孽缘
“”“
在使用Python中进行编程时，程序员无需参与内存的管理工作，这是因为Python有自带的内存管理机制，简称GC。那么GC与GIL有什么关联？

要搞清楚这个问题，需先了解GC的工作原理，Python中内存管理使用的是引用计数，每个数会被加上一个整型的计数器，表示这个数据被引用的次数，当这个整数变为0时则表示该数据已经没有人使用，成了垃圾数据。

当内存占用达到某个阈值时，GC会将其他线程挂起，然后执行垃圾清理操作，垃圾清理也是一串代码，也就需要一条线程来执行。

GC与其他线程都在竞争解释器的执行权，而CPU何时切换，以及切换到哪个线程都是无法预支的，这样一来就造成了竞争问题，假设线程1正在定义变量a=10，而定义变量第一步会先到到内存中申请空间把10存进去，第二步将10的内存地址与变量名a进行绑定，如果在执行完第一步后，CPU切换到了GC线程，GC线程发现10的地址引用计数为0则将其当成垃圾进行了清理，等CPU再次切换到线程1时，刚刚保存的数据10已经被清理掉了，导致无法正常定义变量。

当然其他一些涉及到内存的操作同样可能产生问题问题，为了避免GC与其他线程竞争解释器带来的问题，CPython简单粗暴的给解释器加了互斥锁
”“”

#3、GIL的加锁与解锁时机
“”“
加锁: 只有有一个线程要使用解释器就立马枷锁  

释放:

​    该线程任务结束

​    该线程遇到IO

​    该线程使用解释器过长    默认100纳秒
”“”
#4、总结:
“”“
在CPython中,有了GIL后
    优点：多个线程将不可能在同一时间使用解释器，从而保证了解释器的数据安全
    缺点：GIL的特性使得多线程无法并行，会把线程的并行变成串行,导致效率降低
”“”
#5、解决方案：区分任务类型
“”“
1.单核下无论是IO密集还是计算密集GIL都不会产生任何影响

2.多核下对于IO密集任务，GIL会有细微的影响，基本可以忽略

3.Cpython中IO密集任务应该采用多线程，计算密集型应该采用多进程
”“”
#计算密集型的效率测试
“”“
from multiprocessing import Process
from threading import Thread
import time

def task():
    for i  in range(10000000):
        i += 1

if __name__ == '__main__':
    start_time = time.time()
    # 多进程
    # p1 = Process(target=task)
    # p2 = Process(target=task)
    # p3 = Process(target=task)
    # p4 = Process(target=task)

    # 多线程
    p1 = Thread(target=task)
    p2 = Thread(target=task)
    p3 = Thread(target=task)
    p4 = Thread(target=task)

    p1.start()
    p2.start()
    p3.start()
    p4.start()

    p1.join()
    p2.join()
    p3.join()
    p4.join()
    
    print(time.time()-start_time)
”“”
#IO密集型的效率测试
“”“
from multiprocessing import Process
from threading import Thread
import time
def task():
    with open("test.txt",encoding="utf-8") as f:
        f.read()
if __name__ == '__main__':
    start_time = time.time()
    # 多进程
    # p1 = Process(target=task)
    # p2 = Process(target=task)
    # p3 = Process(target=task)
    # p4 = Process(target=task)

    # 多线程
    p1 = Thread(target=task)
    p2 = Thread(target=task)
    p3 = Thread(target=task)
    p4 = Thread(target=task)

    p1.start()
    p2.start()
    p3.start()
    p4.start()

    p1.join()
    p2.join()
    p3.join()
    p4.join()

    print(time.time()-start_time)
”“”
#6、自定义的线程锁与GIL的区别
“”“
GIL保护的是解释器级别的数据安全，比如对象的引用计数，垃圾分代数据等等，具体参考垃圾回收机制详解。

对于程序中自己定义的数据则没有任何的保护效果，这一点在没有介绍GIL前我们就已经知道了，所以当程序中出现了共享自定义的数据时就要自己加锁
”“”
#自定义锁示例
“”“
from threading import Thread,Lock
import time

lock = Lock()
a = 0
def task():
    global a
    lock.acquire()
    temp = a
    time.sleep(0.01)
    a = temp + 1
    lock.release() 

    
t1 = Thread(target=task)
t2 = Thread(target=task)

t1.start()
t2.start()

t1.join()
t2.join()
print(a)

过程分析：

1.线程1获得CPU执行权，并获取GIL锁执行代码 ，得到a的值为0后进入睡眠，释放CPU并释放GIL，不释放lock

2.线程2获得CPU执行权，并获取GIL锁,尝试获取lock失败，无法执行，释放CPU并释放GIL

3.线程1睡醒后获得CPU执行权，并获取GIL继续执行代码 ，将temp的值0+1后赋给a，执行完毕释放CPU释放GIL，释放lock，此时a的值为1

4.线程2获得CPU执行权，获取GIL锁,尝试获取lock成功，执行代码，得到a的值为1后进入睡眠，释放CPU并释放GIL，不释放lock

5.线程2睡醒后获得CPU执行权，获取GIL继续执行代码 ，将temp的值1+1后赋给a，执行完毕释放CPU释放GIL，释放lock，此时a的值为2


”“”

 　　##线程池与进程池

#线程池和进程池理论基础
#1、什么是进程/线程池？
池表示一个容器，本质上就是一个存储进程或线程的列表

#2、池子中存储线程还是进程？
如果是IO密集型任务使用线程池，如果是计算密集任务则使用进程池

#3、为什么需要进程/线程池？
“”“
1.自动管理线程的开启和销毁 

​2.自动分配任务给空闲的线程

​3.可以线程开启线程的数量 保证系统稳定     

​    信号量中是限制同时并发多少,但是线程已经全都建完了 
”“”
#4、如何使用
“”“
1.创建池子

​2.submit 提交任务 

​3.pool.shutdown() # 等待所有任务全部完毕 销毁所有线程 后关闭线程池

​    关闭后就不能提交新任务了 
”“”

#示例1：多线程对比多进程
"""
io任务
"""
# from multiprocessing import Process
# from threading import Thread
# import time
# def task():
#     time.sleep(2)
#
# if __name__ == '__main__':
#     start = time.time()
#
#     ps = []
#     for i in range(100):
#         p = Process(target=task)
#         p.start()
#         ps.append(p)
#
#     for p in ps:
#         p.join()
#     print(time.time() - start)

# "对于io'密集型 使用多进程反而可能效率比不上多线程 "

"""
计算密集型
"""
#
#
# from multiprocessing import Process
# from threading import Thread
# import time
#
# def task():
#     for i in range(100000000):
#         1 * 10
#
#
# if __name__ == '__main__':
#     start = time.time()
#     ps = []
#     for i in range(5):
#         p = Thread(target=task)
#         p.start()
#         ps.append(p)
#
#     for p in ps:
#         p.join()
#     print(time.time() - start)

"""
对于计算密集型任务而言  应该使用多进程
注意:并行的任务不能太多 开启进程非常消耗资源  
"""


# from threading import Thread,Lock
# import time
#
# a = 0
# def task():
#     global a
#     temp = a
#     time.sleep(0.01)
#     a = temp + 1
#
# t1 = Thread(target=task)
# t2 = Thread(target=task)
# t1.start()
# t2.start()
#
#
# t1.join()
# t2.join()
# print(a)

#示例2：线程的常用方法
“”“
from threading import Thread,current_thread,enumerate,active_count

# 获取当前线程对象
# print(current_thread())
# print(current_thread().__class__)
#
# def task():
#     print(current_thread())
#     print(current_thread().__class__)
#
# Thread(target=task).start()

# 获取正常运行的所有线程对象
# print(enumerate())
import time

Thread(target=lambda : time.sleep(100)).start()

# 获取存活的线程数量
print(active_count())

# 获取线程的名称   其他属性进程对象差不多
print(current_thread().getName())
”“”

#示例3：线程池和进程池
“”“
import os,time
# 获取CPU核心数
# print(os.cpu_count())

"""
    池 Pool 指得是一个容器
    线程池就是用来存储线程对象的 容器
"""
from concurrent.futures import  ThreadPoolExecutor,ProcessPoolExecutor
from threading import enumerate,current_thread

# 1.创建池子 可以指定池子里有多少线程    如果不指定默认为CPU个数 * 5
# 不会立即开启线程 会等到有任务提交后在开启线程
# pool = ThreadPoolExecutor(10)   # 线程池最大值,应该机器所能承受的最大值   当然需要考虑你的机器有几个任务要做
#
# # print(enumerate())
# def task(name,age):
#     print(name)
#     print(current_thread().name,"run")
#     time.sleep(2)
# #
#
# # 该函数提交任务到线程池中
# pool.submit(task,"jerry",10) #任务的参数  直接写到后面不需要定义参数名称   因为是可变位置参数
# pool.submit(task,"owen",20)
# pool.submit(task)
#
# time.sleep(5)
#
# #
# print(enumerate())
"""
    线程池,不仅帮我们管理了线程的开启和销毁,还帮我们管理任务的分配
    特点: 线程池中的线程只要开启之后 即使任务结束也不会立即结束  因为后续可能会有新任务
          避免了频繁开启和销毁线程造成的资源浪费
    1.创建一个线程池 
    2.使用submit提交任务到池子中   ,线程池会自己为任务分配线程  
    
    
"""

# 进程池的使用   同样可以设置最大进程数量 默认为cpu的个数
pool = ProcessPoolExecutor() # 具体的值也要参考机器性能

def task(name):
    print(os.getpid())
    print(name)

if __name__ == '__main__':
    pool.submit(task,"jerry")

    pool.shutdown()

    pool.submit(task, "jerry")
”““

#示例4：线程池的shutdown方法
”“”
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
from threading import  current_thread,enumerate

import time

pool = ThreadPoolExecutor(3)

def task():
    print(current_thread().name)
    print(current_thread().isDaemon())
    time.sleep(1)

for i in range(5):
    pool.submit(task)

st = time.time()

pool.shutdown() # 等待所有任务全部完毕 销毁所有线程 后关闭线程池
# pool.submit(task)

print(time.time() - st)

print("over")

“”“

　　##