并发编程(线程)——验证GIL锁,GIL与普通互斥锁的区别,io密集型和计算密集型,死锁现象(解决方式:递归锁),Semaphore信号量,Event事件,线程queue,多进程实现tcp服务端并发,线程池&进程池

一、验证GIL锁的存在方式

GIL锁:全局解释器锁

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()

二、GIL与普通互斥锁的区别

区别: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)

三、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)

四、死锁现象(哲学家就餐问题)

是指两个或两个以上的进程或线程在执行过程中,因争夺资源而造成的一种互相等待的现象,若无外力作用,它们都将无法推进下去。此时称系统处于死锁状态或系统产生了死锁,这些永远在互相等待的进程称为死锁进程,如下就是死锁
死锁现象:
(1)A线程拿到了A锁,等B锁,B线程拿到了B锁,等A锁,相互等待,永远等下去;
(2)A线程拿到了A锁,再去拿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()

解决方式:

递归锁

递归锁(可重入锁),当前线程可以多次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()

五、Semaphore信号量

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()

六、Event事件

一些线程需要等到其他线程执行完成之后才能执行,类似于发射信号
比如一个线程等待另一个线程执行结束再继续执行
event.set() 发信号
event.wait()阻塞等待信号,只要收到set信号,就会继续往下执行

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()  # 阻塞等信号,只要收到set信号,就会继续往下执行。只要没来信号,就卡在者
    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()

七、线程queue

1、进程queue和线程不是一个
  进程queue:
  from multiprocessing import Queue

  线程queue:
  from queue import Queue,LifoQueue,PriorityQueue

2、不同线程数据交互(线程间通信)的两种方式:
(1)共享变量:因为会出现数据不安全问题,所以不同线程修改同一份数据要加锁(互斥锁)
(2)用线程queue通信:不需要考虑数据安全问题(queue是线程安全的)不需要加锁,内部自带
3、三种线程Queue
-Queue:队列,先进先出
-PriorityQueue:优先级队列,谁小谁先出
-LifoQueue:栈,后进后出
#1 Quenue使用

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())

#2 LifoQueue使用

q=LifoQueue(5)
q.put("lqz")
q.put("egon")
q.put("铁蛋")
q.put("钢弹")
q.put("金蛋")

# q.put("ddd蛋")
print(q.get())

#3 PriorityQueue使用
#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())

八、通过多进程,实现TCP服务端支持多个客户端连接(并发)

#服务端
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()
#客户端
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)

 

九、线程池&进程池

池:池子,用来做缓冲
1、为什么会出现池?不管是开进程还是开线程,不能无限制开,通过池,假设池子里就有10个,不管再怎么开,永远是这10个

2、使用(需要记住)
from concurrent.futures import ThreadPoolExecutor
pool = ThreadPoolExecutor(2)
pool.submit(函数名,参数1,参数2...).add_done_callback(call_back回调函数)
#函数执行完的数据如何给回调函数?
  回调函数会接收一个f对象,对象中有要的数据(函数的return结果),f.result()
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)

线程池和进程池的shutdown(二者用法一样)

#主线程等待所有任务执行完成

from concurrent.futures import ThreadPoolExecutor
import time

pool = ThreadPoolExecutor(3)


def task(name):
    print('%s 开始'%name)
    time.sleep(1)
    print('%s 结束'%name)


if __name__ == '__main__':
    for i in range(20):
        pool.submit(task, '屌丝%s' % i)

    # 放到for外面,等待所有任务执行完成,主线程再继续走
    pool.shutdown(wait=True)  # 等待所有任务完成,并且把池关闭
    #  问题,关了还能提交任务吗?不能再提交了
    # pool.submit(task,'sdddd')#如果关闭后提交数据会报错
    print('') # 立马执行,20个线程都执行完了,再执行

 线程池案例:

  爬网站

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)

十、

---39----

原文地址:https://www.cnblogs.com/guojieying/p/13565776.html