并发编程之锁、进程池、线程池等相关内容-39

1.验证GIL锁

from threading import Thread
from multiprocessing import Process
​
​
def task():
    while True:
        pass
​
​
if __name__ == '__main__':
    for i in range(6):
        # t=Thread(target=task)  # 因为有GIL锁，同一时刻，只有一条线程执行，所以cpu不会满
        t = Process(target=task)  # 由于是多进程，进程中的线程会被cpu调度执行，6个cpu全在工作，就会跑满
        t.start()
​

2.GIL锁和普通互斥锁

from threading import Thread, Lock
import time
​
mutex = Lock()
money = 100
​
​
def task():
    global money
    mutex.acquire()
    temp = money
    time.sleep(1)
    money = temp - 1
    mutex.release()
​
​
if __name__ == '__main__':
    ll = []
    for i in range(10):
        t = Thread(target=task)
        t.start()
        # t.join()  # 会怎么样？变成了串行，不能这么做
        ll.append(t)
    for t in ll:
        t.join()
    print(money)
​

3.IO密集型和计算密集型

'''
​
-----以下只针对于cpython解释器
-在单核情况下：
-开多线程还是开多进程？不管干什么都是开线程
-在多核情况下：
-如果是计算密集型，需要开进程，能被多个cpu调度执行
-如果是io密集型，需要开线程，cpu遇到io会切换到其他线程执行
​
'''
​
from threading import Thread
from multiprocessing import Process
import time
​
​
# 计算密集型
# def task():
#     count = 0
#     for i in range(100000000):
#         count += i
#
#
# if __name__ == '__main__':
#     ctime = time.time()
#     ll = []
#     for i in range(10):
#         t = Thread(target=task)  # 开线程：42.68658709526062
#         # t = Process(target=task)   # 开进程：9.04949426651001
#         t.start()
#         ll.append(t)
#
#     for t in ll:
#         t.join()
#     print(time.time()-ctime)
​
​
## io密集型
def task():
    time.sleep(2)
​
​
if __name__ == '__main__':
    ctime = time.time()
    ll = []
    for i in range(400):
        t = Thread(target=task)  # 开线程：2.0559656620025635
        # t = Process(target=task)   # 开进程：9.506720781326294
        t.start()
        ll.append(t)
​
    for t in ll:
        t.join()
    print(time.time() - ctime)
​

4.死锁现象

# 死锁现象，张三拿到了A锁，等B锁，李四拿到了B锁，等A锁
from threading import Thread, Lock
import time
​
mutexA = Lock()
mutexB = Lock()
​
​
def eat_apple(name):
    mutexA.acquire()
    print('%s 获取到了a锁' % name)
    mutexB.acquire()
    print('%s 获取到了b锁' % name)
    print('开始吃苹果，并且吃完了')
    mutexB.release()
    print('%s 释放了b锁' % name)
    mutexA.release()
    print('%s 释放了a锁' % name)
​
​
def eat_egg(name):
    mutexB.acquire()
    print('%s 获取到了b锁' % name)
    time.sleep(2)
    mutexA.acquire()
    print('%s 获取到了a锁' % name)
    print('开始吃鸡蛋，并且吃完了')
    mutexA.release()
    print('%s 释放了a锁' % name)
    mutexB.release()
    print('%s 释放了b锁' % name)
​
​
if __name__ == '__main__':
    ll = ['egon', 'alex', '铁蛋']
    for name in ll:
        t1 = Thread(target=eat_apple, args=(name,))
        t2 = Thread(target=eat_egg, args=(name,))
        t1.start()
        t2.start()
​

5.递归锁

# 递归锁(可重入)，同一个人可以多次acquire，每acquire一次，内部计数器加1，每relaese一次，内部计数器减一
# 只有计数器不为0，其他人都不获得这把锁
​
from threading import Thread, Lock, RLock
import time
​
# 同一把锁
# mutexA = Lock()
# mutexB = mutexA
​
# 使用可重入锁解决（同一把锁）
# mutexA = RLock()
# mutexB = mutexA
mutexA = mutexB = RLock()
​
​
def eat_apple(name):
    mutexA.acquire()
    print('%s 获取到了a锁' % name)
    mutexB.acquire()
    print('%s 获取到了b锁' % name)
    print('开始吃苹果，并且吃完了')
    mutexB.release()
    print('%s 释放了b锁' % name)
    mutexA.release()
    print('%s 释放了a锁' % name)
​
​
def eat_egg(name):
    mutexB.acquire()
    print('%s 获取到了b锁' % name)
    time.sleep(2)
    mutexA.acquire()
    print('%s 获取到了a锁' % name)
    print('开始吃鸡蛋，并且吃完了')
    mutexA.release()
    print('%s 释放了a锁' % name)
    mutexB.release()
    print('%s 释放了b锁' % name)
​
​
if __name__ == '__main__':
    ll = ['egon', 'alex', '铁蛋']
    for name in ll:
        t1 = Thread(target=eat_apple, args=(name,))
        t2 = Thread(target=eat_egg, args=(name,))
        t1.start()
        t2.start()
​

6.信号量

# Semaphore:信号量可以理解为多把锁，同时允许多个线程来更改数据
​
from threading import Thread, Semaphore
import time
import random
​
sm = Semaphore(3)  # 数字表示可以同时有多少个线程操作
​
​
def task(name):
    sm.acquire()
    print('%s 正在蹲坑' % name)
    time.sleep(random.randint(1, 5))
    sm.release()
​
​
if __name__ == '__main__':
    for i in range(20):
        t = Thread(target=task, args=('屌丝男%s号' % i,))
        t.start()
​

7.EVENT事件

# 一些线程需要等到其他线程执行完成之后才能执行，类似于发射信号
# 比如一个线程等待另一个线程执行结束再继续执行
​
# from threading import Thread, Event
# import time
#
# event = Event()
#
#
# def girl(name):
#     print('%s 现在不单身，正在谈恋爱'%name)
#     time.sleep(10)
#     print('%s 分手了，给屌丝男发了信号'%name)
#     event.set()
#
#
# def boy(name):
#     print('%s 在等着女孩分手'%name)
#     event.wait()  # 只要没来信号，就卡在者
#     print('女孩分手了，机会来了，冲啊')
#
#
# if __name__ == '__main__':
#     lyf = Thread(target=girl, args=('刘亦菲',))
#     lyf.start()
#
#     for i in range(10):
#         b = Thread(target=boy, args=('屌丝男%s号' % i,))
#         b.start()
​
​
# 作业：起两个线程，第一个线程读文件的前半部分，读完发一个信号，另一个进程读后半部分，并打印
​
from threading import Thread, Event
import time
import os
​
event = Event()
# 获取文件总大小
size = os.path.getsize('a.txt')
​
​
def read_first():
    with open('a.txt', 'r', encoding='utf-8') as f:
        n = size // 2  # 取文件一半，整除
        data = f.read(n)
        print(data)
        print('我一半读完了，发了个信号')
        event.set()
​
​
def read_last():
    event.wait()  # 等着发信号
    with open('a.txt', 'r', encoding='utf-8') as f:
        n = size // 2  # 取文件一半，整除
        # 光标从文件开头开始，移动了n个字节，移动到文件一半
        f.seek(n, 0)
        data = f.read()
        print(data)
​
​
if __name__ == '__main__':
    t1 = Thread(target=read_first)
    t1.start()
    t2 = Thread(target=read_last)
    t2.start()
​

8.线程QUEUE

# 进程queue和线程不是一个
# from multiprocessing import Queue
​
# 线程queue
from queue import Queue, LifoQueue, PriorityQueue
​
# 线程间通信，因为共享变量会出现数据不安全问题，用线程queue通信，不需要加锁，内部自带
# queue是线程安全的
​
​
'''
三种线程Queue
    -Queue:队列，先进先出
    -PriorityQueue：优先级队列，谁小谁先出
    -LifoQueue：栈，后进先出
'''
# 如何使用
# q=Queue(5)
# q.put("lqz")
# q.put("egon")
# q.put("铁蛋")
# q.put("钢弹")
# q.put("金蛋")
#
#
# # q.put("银蛋")
# # q.put_nowait("银蛋")
# # 取值
# print(q.get())
# print(q.get())
# print(q.get())
# print(q.get())
# print(q.get())
# # 卡住
# # print(q.get())
# # q.get_nowait()
# # 是否满，是否空
# print(q.full())
# print(q.empty())
​
# LifoQueue
​
# q=LifoQueue(5)
# q.put("lqz")
# q.put("egon")
# q.put("铁蛋")
# q.put("钢弹")
# q.put("金蛋")
# #
# # q.put("ddd蛋")
# print(q.get())
​
​
# PriorityQueue:数字越小，级别越高
​
# q=PriorityQueue(3)
# q.put((-10,'金蛋'))
# q.put((100,'银蛋'))
# q.put((101,'铁蛋'))
# # q.put((1010,'铁dd蛋'))  # 不能再放了
#
# print(q.get())
# print(q.get())
# print(q.get())
​

9.并发的TCP通信

# server.py

from multiprocessing import Process

import socket


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


if __name__ == '__main__':
    server = socket.socket()

    server.bind(('127.0.0.1', 8081))
    server.listen(5)

    # 多线程，或者多进程
    while True:  # 连接循环
        conn, addr = server.accept()
        # 多用户的服务端
        t = Process(target=task, args=(conn,))
        t.start()

        ### 单用户的服务端
        # while True:
        #     try:
        #         data = conn.recv(1024)
        #         if len(data) == 0: break
        #         print(data)
        #         conn.send(data.upper())
        #     except Exception as e:
        #         print(e)
        #         break
        # conn.close()

        
# client.py

import socket

import time

cli = socket.socket()
cli.connect(('127.0.0.1', 8081))

while True:
    cli.send(b'hello world')
    time.sleep(0.1)
    data = cli.recv(1024)
    print(data)

10.线程池小案例

from concurrent.futures import ThreadPoolExecutor

import requests  # 爬虫会学到的模块

pool = ThreadPoolExecutor(2)


def get_pages(url):
    # https://www.baidu.com
    res = requests.get(url)  # 向这个地址发送请求

    name = url.rsplit('/')[-1] + '.html'
    print(name)  # www.baidu.com.html
    # res.content拿到页面的二进制
    return {'name': name, 'text': res.content}


def call_back(f):
    dic = f.result()
    with open(dic['name'], 'wb') as f:
        f.write(dic['text'])


if __name__ == '__main__':
    ll = ['https://www.baidu.com', 'https://www.mzitu.com', 'https://www.cnblogs.com']
    for url in ll:
        pool.submit(get_pages, url).add_done_callback(call_back)

11.线程池和进程池

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
from threading import Thread
import time
import random

pool = ThreadPoolExecutor(5)  # 数字是池的大小


# pool = ProcessPoolExecutor(5)  # 数字是池的大小


def task(name):
    print('%s任务开始' % name)

    time.sleep(random.randint(1, 4))
    print('任务结束')
    return '%s 返回了' % name


def call_back(f):
    # print(type(f))
    print(f.result())


if __name__ == '__main__':

    # ll=[]
    # for i in range(10):  # 起了100个线程
    #     # t=Thread(target=task)
    #     # t.start()
    #     res = pool.submit(task, '屌丝男%s号' % i)  # 不需要再写在args中了
    #     # res是Future对象
    #     # from  concurrent.futures._base import Future
    #     # print(type(res))
    #     # print(res.result())  # 像join,只要执行result，就会等着结果回来，就变成串行了
    #     ll.append(res)
    #
    # for res in ll:
    #     print(res.result())

    # 终极使用
    for i in range(10):  # 起了100个线程
        # 向线程池中提交一个任务，等任务执行完成，自动回到到call_back函数执行
        pool.submit(task, '屌丝男%s号' % i).add_done_callback(call_back)