python之并发编程

一、操作系统与应用程序

　　在没有介绍线程与进程相关的知识之前，需要给大家引入一个知识点，关于操作系统与应用程序又与线程与进程有什么关系？

　　硬件
    ...
　　装系统
    　　系统就是一个有程序员写出来的软件,该软件用于控制计算机硬件,让他们之间相互配合
　　装软件(安装应用程序)
    　　...
　　并发与并行
    　　并发,假,由于执行速度贴别快,感觉不到停顿
    　　并行,真,创建10个人同时操作
　　线程与进程
    　　单线程与单进程
        　　一个软件默认一个进程,可以有多个
    　　线程:
        　　工作的最小单元
        　　共享进程中的所有资源
        　　每个线程可以分担一些任务，最终完成最后的结果
    　　进程:
        　　独立开辟内存
        　　进程之间的数据隔离
    　　注:GIL锁,是在进程中起到了阻塞线程的作用
　　总结:
    　　1.操作系统帮助开发者操作硬件
    　　2.程序员写好代码在操作系统上运行(依赖解释器)
    　　3.任务特别多
       　　 python多线程情况下:
            　　计算密集型操作:效率低
            　　IO操作:效率高
        　　python多进程情况下:
            　　计算密集型操作:效率高(浪费资源)
            　　IO操作:效率高(浪费资源)
        　　以后写python时:
            　　IO密集型用多线程
           　　计算密集型用多进程
        　　java多线程情况下:
            　　计算密集型操作:效率高
            　　IO操作:效率高
        　　java多进程情况下:
            　　计算密集型操作:效率高(浪费资源)
            　　IO操作:效率高(浪费资源)

    GIL锁:全局解释器锁,用于限制一个进程中间同一个时刻只有一个线程被cpu调度
    扩展:默认GIL锁在执行100个指令

二、线程/进程/协程
　　线程是程序中一个单一的顺序控制流程。进程内有一个相对独立的、可调度的执行单元，是系统独立调度和分派CPU的基本单位
　指令运行时的程序的调度单位。在单个程序中同时运行多个线程完成不同的工作，称为多进程。
　　
　　线程的简单示例:
　　　　

import time
    import threading
    def task():
        time.sleep(0)   #0秒也会有延迟
        print("拿快递")
    # 创建一个线程
    t = threading.Thread(target=task)
    # 执行线程
    t.start()

    print("干什么。。。")

    import time
    import threading

    def task(n):
        print("执行任务",n)
        time.sleep(5)
        print("...")
        print("任务%s 执行完毕" %n)
    while True:
        name = input("请输入任务")
        t = threading.Thread(target=task,args=(name,))
        t.start()

所以说线程就像是流水线一样，走到哪里该作什么，是早已经规划好的。

　　线程的使用:

　　

import threading,time
def num():
    print("456")
t = threading.Thread(target=num)
t.start()
print("123")
# 运行结果
456
123

　　setDaemon方法

　　

import threading,time
def sleep(arg):
    time.sleep(arg)
    print(arg)

t1 = threading.Thread(target=sleep,args=(3,))
t1.start()

t2 = threading.Thread(target=sleep,args=(6,))
t2.setDaemon(True)
t2.start()

print("1")
# 运行结果
# 1
# 3
# 总结，线程中的主线程默认会等其他线程执行完毕后才关闭
# 当设置setDaemon(True)的时候会不等

　　join方法

　　

import threading,time

def join(arg):
    time.sleep(arg)
    print(arg)
print("123")
t1 = threading.Thread(target=join,args=(3,))
t1.start()
t1.join()
print("456")
#  运行结果:
# 123
# 3
# 456
# 总结:当在子线程中添加join方法(可以加参数，表示最多等多久)，则会等待子线程运行完毕后执行主线程

　　获取线程的名称相关

　　

import threading
def name():
    t = threading.current_thread() # 获取当前线程的对象
    print(t.name)

t1 = threading.Thread(target=name)
t1.setName("amazing")
t1.start()

t2 = threading.Thread(target=name)
t2.setName("Magic")
t2.start()
# 运行结果
# amazing
# Magic
# 总结:可以在函数中添加threading.current_thread()用来获取当前线程的对象

　　线程池

from concurrent.futures import ThreadPoolExecutor
import time

pool = ThreadPoolExecutor(5)        #创建一个线程池，并设置同时最多有5个线程可以被调度
def func(arg):
    time.sleep(1)
    print(arg)
for i in range(4):
    pool.submit(func,i)      # 去线程池中申请一个线程执行func函数
# 有节制的约束线程的创建，保证线程的性能。

 　　进程是计算机是系统进行资源分配和调度的基本单位，进程是线程的容器。程序是指令、数据及其组织形式的描述，进程是程序的实体。

 　　进程的编写

import multiprocessing

def task(arg):
    print(arg)
def run():
    for i in range(10):
        p = multiprocessing.Process(target=task,args=(i,))
        p.start()
if __name__ == '__main__':    # 在windows系统中 需要将主进程放在main方法下运行
    run()

　　进程的使用

　　join

import multiprocessing,time
def task(arg):
    time.sleep(2)
    print(arg)
def run():
    for i in range(10):
        p = multiprocessing.Process(target=task,args=(i,))
        p.start()
        p.join(1)
        print("等待结束")
if __name__ == '__main__':
    run()
# 总结:join等待进程完成过后才执行

　　daemon

import multiprocessing,time
def task(arg):
    time.sleep(2)
    print(arg)
def run():
    print("123")
    p = multiprocessing.Process(target=task,args=(789,))
    p.daemon = True
    p.start()

    print("456")
if __name__ == '__main__':
    run()
总结:当daemon=True表示不等待

　　name

import multiprocessing,time
def task(arg):
    p = multiprocessing.current_process()
    print(p.pid)  #p.ident
    time.sleep(2)
    print(arg)
def run():
    for i in range(10):
        p = multiprocessing.Process(target=task,args=(i,))
        p.name = "asd"
        p.start()
        p.join(1)
        print("等待结束")
if __name__ == '__main__':
    run()

　　进程间的相关问题

　　　　进程间的数据是否相互共享?

import multiprocessing
data_list =[]
def task(arg):
    data_list.append(arg)
    print(data_list)
def run():
    for i in range(10):
        p = multiprocessing.Process(target=task,args=(i,))
        p.start()
if __name__ == '__main__':
    run()
进程中的数据是不共享的

　　　　怎么做才能够数据间的相互共享

　　　　　　Queue()

　　　　

import multiprocessing,queue,time

q = multiprocessing.Queue()
def task(arg):
    q.put(arg)

def run():
    for i in range(10):
        p = multiprocessing.Process(target=task,args=(i,))
        p.start()
    time.sleep(2)
    v1 = q.get()
    print(v1)
if __name__ == '__main__':
    run()

　　　　　　Manager()

import multiprocessing,queue,time
m = multiprocessing.Manager()
dic = m.dict()

def task(arg):
    dic[arg] = 100

def run():
    for i in range(10):
        p = multiprocessing.Process(target=task,args=(i,))
        p.start()
        p.join()
    input(">>>")
    print(dic.valus)
if __name__ == '__main__':
    run()

　　进程池

from concurrent.futures import ProcessPoolExecutor
import time

def task(arg):
    time.sleep(2)
    print(arg)

if __name__ == '__main__':
    pool  = ProcessPoolExecutor(5)
    for i in range(10):
        pool.submit(task,i)

 　　协程不是真实存在的东西,是程序员创造出来的，协程指的是微线程。

　　协程的编写

import greenlet
def func1():
    print("123")
    gr2.switch()
    print("987")
    gr2.switch()
def func2():
    print("465")
    gr1.switch()
    print("789")
# 协程1
gr1 = greenlet.greenlet(func1)
# 协程2
gr2 = greenlet.greenlet(func2)

# 执行协程1
gr1.switch()

　　

基于yield实现协程
def f1():
    print(11)
    yield
    print(22)
    yield
    print(33)

def f2():
    print(44)
    yield
    print(55)
    yield
    print(66)
v1 = f1()
v2 = f2()
next(v1)

　　协程的作用

　　对一个线程进程分片，使得线程在代码块之间进行来回切换执行，而不是按原来的流程执行。
　　...

三、锁相关

　　"锁"顾名思义，就是用来约束某种对象。

　　锁的简单示例

import threading,time
lock = threading.RLock()  # 创建锁都对象
n = 10
def lk(i):
    print("代码不加锁",i)
    lock.acquire()  #加锁
    global n
    print("当前线程",i,"读取到的n值为:",n)
    n = i
    time.sleep(1)
    print("当前线程",i,"读取到的n值为:",n)
    lock.release()  #释放锁头
for i in range(10):
    t = threading.Thread(target=lk,args=(i,))
    t.start()
# 总结:加完锁之后此区域的代码同一时刻只能有一个线程执行

线程安全与线程不安全的问题
    当多线程操作时，内不会让所有线程排队处理这时候线程是安全的，例如:list/dict/Queue
    线程不安全 + 人(锁) => 排队处理

锁的用法:
　　lock 普通锁　最简单的锁

import threading,time
v = []
lock = threading.Lock()    # 创建一个lock对象
def func(arg):
    lock.acquire()      # 添加锁
    v.append(arg)
    time.sleep(1)
    m = v[-1]
    print(arg,m)
    lock.release()          # 释放锁
for i in range(10):
    t = threading.Thread(target=func,args=(i,))
    t.start()
# 总结:在普通锁中当多个lock存在时会死锁

　　Rlock 递归锁 也是最常用的锁

import threading,time
lock = threading.RLock()
v = []
def func(arg):
    lock.acquire()
    lock.acquire()      # 添加多个锁
    v.append(arg)
    time.sleep(1)
    m = v[-1]
    print(arg,m)
    lock.release()
    lock.release()
for i in range(10):
    t = threading.Thread(target=func,args=(i,))
    t.start()

# 总结:支持递归的锁多次解多次，允许多个lock同时存

　　BoundedSemapore 信号锁

import threading,time
lock = threading.BoundedSemaphore(5)     # 表示通过设定可以放行n个线程

def func(arg):
    lock.acquire()
    print(arg)
    time.sleep(1)
    lock.release()
for i in range(20):
    t = threading.Thread(target=func,args=(i,))
    t.start()

# 总结:可以一次放行多个线程。

　　Condition 条件锁

import threading,time
lock = threading.Condition()
def func(arg):
    lock.acquire()
    lock.wait()     # 表示等待,线程到此处停止需要一个接收值通知放行多少个线程
    print(arg)
    time.sleep(1)
    lock.release()
for i in range(20):
    t = threading.Thread(target=func,args=(i,))
    t.start()

while True:
    user_input = int(input("请输入要放行的个数"))
    lock.acquire()
    lock.notify(user_input)
    lock.release()
总结: 在使用该方法时需要在主线程或其他线程中，写通知wait的方法，告诉wait放行多少个线程

　　Event 事件锁

import time,threading
lock = threading.Event()
def func(arg):
    lock.wait()    # 表示当线程到此处停住，相当于红灯
    print(arg)
for i in range(10):
    t = threading.Thread(target=func,args=(i,))
    t.start()
user_input = input("输入红灯或绿灯")
if user_input =="绿灯":
    lock.set()    # 表示一次性释放所有线程 ，相当于绿灯
else:
    pass

# 总结: 一次性放行所有

 　　进程锁的使用方法大体与线程数相同，只是在导入模块时是import multiprocessing，调用的时候也是。

为甚么要用锁?
    1.线程安全,列表和字典线程安全:
        非线程安全
        控制一段代码
　　2.如果多个线程同时修改一个值的时候需要加锁

线程锁/进程锁/GIL锁
　　线程锁是为了在线程进行数据操作时保证数据的安全性(涉及到线程安全问题)
　　进程锁是为了在数据共享时起到防护的作用
　　GIL锁是为了限制同一个进程中只能有一个线程进入python解释器

threading.local()  *

import threading,time
v = threading.local()

def func(arg):
    v.space = arg   # 在函数内部会创建一个属于自己的空间
    time.sleep(1)
    print(v.space,arg)

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

# 总结:为不同的线程创建一个空间(字典)，可以去当前自己空间里面存取值

生产者消费者模型
    生产者
    　　队列:先进先出
    　　栈:先进后出
    消费者
        更便利

　　简单示例:

import queue,threading,time

q = queue.Queue()

def producer(id):
    """
    表示生产者
    :param id:
    :return:
    """
    while True:
        time.sleep(1)
        q.put("发送信息")
        print("%s号用户发送信息"%id)

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

def consumer(id):
    """
    表示消费者
    :param id:
    :return:
    """
    while True:
        time.sleep(1)
        q.get("获得信息")
        print("%s号客服获得信息"%id)

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

# 总结:生产者消费者模型解决了不用一直等待的问题

。。。