Python——并发编程02

Posted 2021-02-21 夕海_Yumi

死锁与递归锁（了解）

当你知道锁的使用抢锁必须要释放锁，其实你在操作锁的时候也极其容易产生死锁现象(整个程序卡死 阻塞)

‘‘‘死锁现象‘‘‘
from threading import Thread, RLock
import time
mutexA = mutexB = RLock() # 递归锁解决死锁现象
# mutexA = Lock()
# mutexB = Lock() # 此时会产生死锁现象，即线程1手里有B锁想要A锁，线程2手里有A锁想要B锁
# 类只要加括号多次 产生的肯定是不同的对象
# 如果你想要实现多次加括号等到的是相同的对象 单例模式
‘‘‘
递归锁：递归锁只能被第一个人抢到，递归锁可以被连续的acquire和release，
	   每acquire()一次其计数+1，
	   每release()一次其计数-1，
	   只要计数不为0，其他人就无法抢到该锁
‘‘‘


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

    def f1(self):
        mutexA.acquire()
        print(‘%s 拿到了A锁‘ % self.name) # 获取当前线程名
        mutexB.acquire()
        print(‘%s 拿到了B锁‘ % self.name)
        mutexB.release()
        mutexA.release()

    def f2(self):
        mutexB.acquire()
        print(‘%s 拿到了B锁‘ % self.name)
        time.sleep(0.1)
        mutexA.acquire()
        print(‘%s 拿到了A锁‘ % self.name)
        mutexA.release()
        mutexB.release()


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

信号量（了解）

信号量在不同的阶段可能对应不同的技术点

在并发编程中信号量指的是锁!!!

from threading import Thread, Semaphore, current_thread
import time,random
# 括号内填数字，如果不填则默认为1
sm = Semaphore(5) # 表示最多只能被五个人抢到


‘‘‘random模块还可以实现打印随机验证码‘‘‘

def task():
    sm.acquire()
    print(‘%s is loading‘ % current_thread().name)
    time.sleep(random.randint(1, 3))
    sm.release()


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

Event事件（了解）

一些进程/线程需要等待另外一些进程/线程运行完毕之后才能运行，类似于发射信号一样

‘‘‘Event事件‘‘‘
from threading import Thread, Event
import time


event = Event()  # 造了一个红绿灯


def light():
    print(‘红灯亮着的‘)
    time.sleep(3)
    print(‘绿灯亮了‘)
    # 告诉等待红灯的人可以走了
    event.set()


def car(name):
    print(‘%s 车正在等红灯‘%name)
    event.wait()  # 等待别人给你发信号
    print(‘%s 车加油门飙车走了‘%name)


if __name__ == ‘__main__‘:
    t = Thread(target=light)
    t.start()

    for i in range(20):
        t = Thread(target=car, args=(‘%s‘%i, ))
        t.start()

‘‘‘定时器‘‘‘
from threading import Timer


def task():
    print(‘hello world‘)


t = Timer(2, task)
t.start()

线程queue（了解）

"""
同一个进程下多个线程数据是共享的
为什么在同一个进程下还会去使用队列呢
因为队列是
    管道 + 锁
所以用队列还是为了保证数据的安全
"""
import queue
# 我们现在使用的队列都是只能在本地测试使用

# 1 队列q  先进先出
q = queue.Queue(3)
q.put(1) # 往队列里添加数据
q.get()	# 往队列里取出数据
q.get_nowait() # 抛出异常，不让队列顿死
q.get(timeout=3) # 设置等待时间，如果超时后还没有获取数据则抛出异常
q.full() # 判断队列是否满
q.empty() # 判断队列是否被取完

# 后进先出q
q = queue.LifoQueue(3)  # last in first out
q.put(1)
q.put(2)
q.put(3)
print(q.get())  # 3

# 优先级q   你可以给放入队列中的数据设置进出的优先级
q = queue.PriorityQueue(4)
q.put((10, ‘111‘))
q.put((100, ‘222‘))
q.put((0, ‘333‘))
q.put((-5, ‘444‘))
print(q.get())  # (-5, ‘444‘)
# put括号内放一个元祖  第一个放数字表示优先级
# 需要注意的是 数字越小优先级越高!!!

进程池与线程池（掌握）

进程池与线程池理论篇

‘‘‘
无论是开设进程还是线程，都需要消耗资源，只不过开设线程耗资要比进程小一点而已

我们不可能无限制的开设进程和线程，因为软件的运行归根到底还是使用硬件的资源，
而我们硬件的开发速度远远赶不上软件

总结：必须在保证硬件能够正常工作的情况下最大限度的利用它
‘‘‘
# 池的概念
‘‘‘
能够在保证硬件安全的情况下最大限度的利用计算机，虽然降低了程序运行效率但是保证了硬件安全
从而使得我们的程序能够正常运行
‘‘‘

进程池与线程池操作篇

# 基本使用
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import time
import os


# pool = ThreadPoolExecutor(5)  # 池子里面固定只有五个线程
# 括号内可以传数字 不传的话默认会开设min(32, (os.cpu_count() or 1) + 4)个的线程
pool = ProcessPoolExecutor(5)
# 括号内可以传数字 不传的话默认会开设当前计算机cpu个数进程
"""
池子造出来之后 里面会固定存在五个线程 # 避免了重复开设线程的资源消耗
这个五个线程不会出现重复创建和销毁的过程
池子造出来之后 里面会固定的几个进程 # 避免了重复开设进程的资源消耗
这个几个进程不会出现重复创建和销毁的过程

池子的使用：只需要将要做的任务提交到池中就可以，会自动有人来服务你
"""


def task(n):
    print(n,os.getpid())
    time.sleep(2)
    return n**n

def call_back(n):
    print(‘call_back>>>:‘,n.result())
"""
任务的提交方式
    同步:提交任务之后原地等待任务的返回结果 期间不做任何事
    异步:提交任务之后不等待任务的返回结果 直接继续往下执行
        返回结果如何获取？？？
        异步提交任务的返回结果 应该通过回调机制来获取
        回调机制
            就相当于给每个异步任务绑定了一个定时炸弹
            一旦该任务有结果立刻触发爆炸
"""
if __name__ == ‘__main__‘:
    t_list = []
    for i in range(10):  # 朝池子中提交10个任务
        # submit()是一个异步提交，返回值是Future类的一个对象
        # res = pool.submit(task, i)  # <Future at 0x100f97b38 state=running>
        res = pool.submit(task, i).add_done_callback(call_back)
        # print(res.result())  # result方法   同步提交， 由并发变成了串行
        # t_list.append(res)
    # 等待线程池中所有的任务执行完毕之后再继续往下执行
    # pool.shutdown()  # 关闭线程池  等待线程池中所有的任务运行完毕
    # for t in t_list:
    #     print(‘>>>:‘,t.result())  # 肯定是有序的
"""
程序由并发变成了串行
任务的为什么打印的是None
res.result() 拿到的就是异步提交的任务的返回结果
"""

总结

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
# pool = ThreadPoolExecutor(5)
pool = ProcessPoolExecutor(5)
pool.submit(task, i).add_done_callback(call_back)

协程（了解）

‘‘‘
进程：资源单位
线程：执行单位
协程：只是一种思想，在单线程下实现并发效果，通过代码层面完成遇到IO就切换，以此来骗过cpu
	 让cpu觉得程序一直在运行且没有IO，从而提升程序的运行效率
	 
代码如何做到切换+保存状态
切换
	切换不一定就提升效率，在IO密集型中会提升效率但是在计算密集型中会降低效率
保存状态
	保存上一次我执行的状态 下一次来接着上一次的操作继续往后执行
	yield
‘‘‘

验证切换是否一定提升效率

import time

# 串行执行计算密集型的任务   1.2372429370880127
def func1():
    for i in range(10000000):
        i + 1

def func2():
    for i in range(10000000):
        i + 1

start_time = time.time()
func1()
func2()
print(time.time() - start_time)

# 切换 + yield  2.1247239112854004
import time


def func1():
    while True:
        10000000 + 1
        yield


def func2():
    g = func1()  # 先初始化出生成器
    for i in range(10000000):
        i + 1
        next(g)

start_time = time.time()
func2()
print(time.time() - start_time)

协程实现TCP服务端的并发效果（了解）

gevent模块

# 安装
‘‘‘
pip3 install gevent
‘‘‘

# 使用
from gevent import monkey;monkey.patch_all()
import time
from gevent import spawn

"""
gevent模块本身无法检测常见的一些io操作
在使用的时候需要额外的导入一句话
from gevent import monkey
monkey.patch_all()
又由于上面的两句话在使用gevent模块的时候是肯定要导入的
所以还支持简写
from gevent import monkey;monkey.patch_all()
"""


def heng():
    print(‘哼‘)
    time.sleep(2)
    print(‘哼‘)


def ha():
    print(‘哈‘)
    time.sleep(3)
    print(‘哈‘)


def heiheihei():
    print(‘heiheihei‘)
    time.sleep(5)
    print(‘heiheihei‘)


start_time = time.time()
g1 = spawn(heng)# 用来监管heng中的IO操作，是异步提交且有返回值
g2 = spawn(ha)# 用来监管ha中的IO操作
g3 = spawn(heiheihei)# 用来监管heiheihei中的IO操作
g1.join()
g2.join()  # 等待被检测的任务执行完毕 再往后继续执行
g3.join()
# heng()
# ha()
# print(time.time() - start_time)  # 5.005702018737793
print(time.time() - start_time)  # 3.004199981689453   5.005439043045044

协程实现TCP服务端的并发

# 服务端
from gevent import monkey;monkey.patch_all()
import socket
from gevent import spawn


def communication(conn):
    while True:
        try:
            data = conn.recv(1024)
            if len(data) == 0: break
            conn.send(data.upper())
        except ConnectionResetError as e:
            print(e)
            break
    conn.close()


def server(ip, port):
    server = socket.socket()
    server.bind((ip, port))
    server.listen(5)
    while True:
        conn, addr = server.accept()
        spawn(communication, conn)


if __name__ == ‘__main__‘:
    g1 = spawn(server, ‘127.0.0.1‘, 8080)
    g1.join()
    
# 客户端
from threading import Thread, current_thread
import socket


def x_client():
    client = socket.socket()
    client.connect((‘127.0.0.1‘,8080))
    n = 0
    while True:
        msg = ‘%s say hello %s‘%(current_thread().name,n)
        n += 1
        client.send(msg.encode(‘utf-8‘))
        data = client.recv(1024)
        print(data.decode(‘utf-8‘))


if __name__ == ‘__main__‘:
    for i in range(500):
        t = Thread(target=x_client)
        t.start()

总结

"""
理想状态:
	我们可以通过
	多进程下面开设多线程
	多线程下面再开设协程序
	从而使我们的程序执行效率提升
"""