并发编程之多线程

Posted 2020-12-03 xfxing

一、什么是多线程？
进程只是用来把资源集中到一起（进程只是一个资源单位，或者说资源集合），而线程才是cpu上的执行单位。
多线程：指在一个进程中存在多个线程，多个线程共享该进程的地址空间。

二、线程和进程的区别
1，线程共享创建它的进程的地址空间;进程有自己的地址空间。
2，线程可以直接访问其进程的数据段;进程有自己的父进程数据段的副本。
3，线程可以直接与进程的其他线程通信;进程必须使用进程间通信来与同胞进程通信。
4，新线程很容易创建;新进程需要父进程的重复。
5，线程可以对相同进程的线程进行相当大的控制;进程只能对子进程进行控制。
6，对主线程的更改(取消、优先级更改等)可能会影响进程的其他线程的行为;对父进程的更改不会影响子进程。

因此我们在特定的场景下需要使用多线程：
1.同一个进程内的多个线程共享该进程内的地址资源
2.创建线程的开销要远小于创建进程的开销（创建一个进程，就是创建一个车间，涉及到申请空间，而且在该空间内建至少一条流水线，但创建线程，就只是在一个车间内造一条流水线，无需申请空间，所以创建开销小）

三、多线程应用举例
开启一个字处理软件进程，该进程肯定需要办不止一件事情，比如监听键盘输入，处理文字，定时自动将文字保存到硬盘，这三个任务操作的都是同一块数据，因而不能用多进程。只能在一个进程里并发地开启三个线程,如果是单线程，那就只能是，键盘输入时，不能处理文字和自动保存，自动保存时又不能输入和处理文字。

四、threading模块介绍
multiprocess模块模仿了threading模块的接口；使用层面上非常相似

五、开启线程的两种方式
方式一：
import time
import random
from threading import Thread

def study(name):
print("%s is learning" % name)
time.sleep(random.randint(1, 3))
print("%s is done" % name)

if name == ‘main‘:
t = Thread(target=study,args=(‘james‘,))
t.start()
print("主线程开始运行....")

方式二：
import time
import random
from threading import Thread

class MyThread(Thread):
def init(self, name):
super().__init__()
self.name = name

def run(self):
    print("%s is learning" % self.name)
    time.sleep(random.randint(1, 3))
    print("%s is done" % self.name)

if name == ‘main‘:
t = MyThread(‘james‘)
t.start()
print("主线程开始运行....")

六、基于多线程实现并发的套接字通信
客户端

_*_ coding: utf-8 _*_
from socket import *

ip_port = (‘127.0.0.1‘,9999)
client = socket(AF_INET,SOCK_STREAM)
client.connect(ip_port)

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

服务端
import multiprocessing
import threading
import socket

ip_port = (‘127.0.0.1‘,9999)
s = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
s.bind(ip_port)
s.listen(5)

def action(conn):
while True:
data = conn.recv(1024)
print(data)
conn.send(data.upper())
if name == ‘main‘:
while True:
conn,addr = s.accept()

    p = threading.Thread(target=action,args=(conn,))
    p.start()

七、多线程与多进程的区别
进程要申请内存空间，开进程的开销远远大于开线程

在主进程下开启线程
import time
import random
from multiprocessing import Process
from threading import Thread

def study(name):
print("%s is learning"%name)
time.sleep(random.randint(1,3))
print("%s is playing" % name)

if name == ‘main‘:
t = Process(target=study,args=(‘james‘,))
t.start()
print("主进程程开始运行....")

执行结果：在t.start()的同时将线程开启了

在主进程下开启子进程
import time
import random
from multiprocessing import Process
from threading import Thread

def study(name):
print("%s is learning"%name)
time.sleep(random.randint(1,3))
print("%s is playing" % name)

if name == ‘main‘:

t = Thread(target=study,args=('james',))
t.start()
print("主线程开始运行....")

执行结果：p.start()将开启进程的信号发给操作系统后，操作系统要申请内存空间，让拷贝好的父进程地址空间到子进程

查看PID
在主线程下开启多个线程，每个线程都跟主线程的pid一样（线程共享主进程的pid）
from threading import Thread
import os

def work():
print(‘hello‘,os.getpid())

if name == ‘main‘:
t1=Thread(target=work)
t2=Thread(target=work)
t1.start()
t2.start()
print(‘主线程/主进程pid‘,os.getpid())

开多个进程，每个进程都有不同的pid
from multiprocessing import Process
import os

def work():
print(‘hello‘,os.getpid())

if name == ‘main‘:
p1=Process(target=work)
p2=Process(target=work)
p1.start()
p2.start()
print(‘主线程/主进程‘,os.getpid())

同一进程内的线程共享该进程的数据
进程之间地址空间是隔离的
from multiprocessing import Process
import os

def work():
global n
n=0

if name == ‘main‘:
n=100
p=Process(target=work)
p.start()
p.join()
print(‘主‘,n)

执行结果：子进程p已经将自己的全局的n改成了0,但改的仅仅是它自己的,查看父进程的n仍然为100

from threading import Thread
import os

def work():
global n
n=0

if name == ‘main‘:
n=100
t=Thread(target=work)
t.start()
t.join()
print(‘主‘,n)

执行结果：查看结果为0,因为同一进程内的线程之间共享进程内的数据

八、Thread对象的其他属性或方法
介绍
Thread实例对象的方法
# isAlive(): 返回线程是否活动的。
# getName(): 返回线程名。
# setName(): 设置线程名。

threading模块提供的一些方法：
# threading.currentThread(): 返回当前的线程变量。
# threading.enumerate(): 返回一个包含正在运行的线程的list。正在运行指线程启动后、
结束前，不包括启动前和终止后的线程。
# threading.activeCount(): 返回正在运行的线程数量，与len(threading.enumerate())有相同的结果。

from threading import Thread
from threading import current_thread
import time

def task():
print("%s is running"%current_thread().getName())
time.sleep(1)
print("%s is done" % current_thread().getName())

if name ==‘main‘:
#没有子线程这个概念，只是为了理解方便
t = Thread(target=task,name=‘子线程1‘)
t.start()
t.setName(‘儿子线程1‘)
print("主线程 %s" % current_thread().getName())
#主线程 MainThread
#子线程1 is running
#儿子线程1 is done

主线程等待子线程结束
from threading import Thread
import time
def sayhi(name):
time.sleep(2)
print(‘%s say hello‘ %name)

if name == ‘main‘:
t=Thread(target=sayhi,args=(‘james‘,))
t.start()
t.join()
print(‘主线程‘)
print(t.is_alive())
‘‘‘
james say hello
主线程
False
‘‘‘

九、守护线程
1 主进程在其代码结束后就已经算运行完毕了（守护进程在此时就被回收）,然后主进程会一直等非守护的子进程都运行完毕后回收子进程的资源(否则会产生僵尸进程)，才会结束。
2 主线程在其他非守护线程运行完毕后才算运行完毕（守护线程在此时就被回收）。因为主线程的结束意味着进程的结束，进程整体的资源都将被回收，而进程必须保证非守护线程都运行完毕后才能结束。

from threading import Thread
import time

def task(name):
time.sleep(1)
print("%s is working"%name)

if name == ‘main‘:
t = Thread(target=task,args=(‘james‘,))
#t.setDaemon(True)#必须在t,start()之前设置，和t.daemon是一样的作用
t.daemon =True
t.start()
print("主线程")
print(t.is_alive())

互斥锁：与进程的互斥锁类似

mutex

from threading import Thread,Lock
import time

n=100

def task():
global n
mutex.acquire()
temp=n
time.sleep(0.1)
n=temp-1
mutex.release()

if name == ‘main‘:
mutex=Lock()
t_l=[]
for i in range(100):
t=Thread(target=task)
t_l.append(t)
t.start()

for t in t_l:
    t.join()

print('主',n)

十、GIL全局解释器锁
GIL本质就是一把互斥锁，既然是互斥锁，所有互斥锁的本质都一样，都是将并发运行变成串行，以此来控制同一时间内共享数据只能被一个任务所修改，进而保证数据安全。
每次执行python程序，都会产生一个独立的进程。

GIL与Lock的区别
锁的目的是为了保护共享的数据，同一时间只能有一个线程来修改共享的数据
保护不同的数据就应该加不同的锁
GIL 与Lock是两把锁，保护的数据不一样，前者是解释器级别的，后者是保护用户自己开发的应用程序的数据，很明显GIL不负责这件事，只能用户自定义加锁处理，即Lock。

from threading import Thread,Lock
import os,time
def work():
global n
lock.acquire()
temp=n
time.sleep(0.1)
n=temp-1
lock.release()
if name == ‘main‘:
lock=Lock()
n=100
l=[]
for i in range(100):
p=Thread(target=work)
l.append(p)
p.start()
for p in l:
p.join()

print(n) #结果肯定为0，由原来的并发执行变成串行，牺牲了执行效率保证了数据安全，不加锁则结果可能为99

GIL与多线程
有了GIL的存在，同一时刻同一进程中只有一个线程被执行
1、对计算来说，cpu越多越好，但是对于I/O来说，再多的cpu也没用（CPU密集型代码）
2、当然对运行一个程序来说，随着cpu的增多执行效率肯定会有所提高（IO密集型代码）

在python多线程下，每个线程的执行方式：
1.获取GIL
2.执行代码直到sleep或者是python虚拟机将其挂起
3.释放GIL

每次释放GIL锁，线程进行锁竞争，切换线程，会消耗资源。并且由于GIL锁存在，python里一个进程永远只能同时执行一个线程（拿到GIL的线程才能执行），这就是为什么在多核CPU上，python的多线程效率并不高的原因。

多线程性能测试
如果并发的多个任务是计算密集型：多进程效率高
from multiprocessing import Process
from threading import Thread
import os,time
def work():
res=0
for i in range(100000000):
res*=i

if name == ‘main‘:
l=[]
print(os.cpu_count()) #本机为4核
start=time.time()
for i in range(4):
p=Process(target=work) #耗时5s多
p=Thread(target=work) #耗时18s多
l.append(p)
p.start()
for p in l:
p.join()
stop=time.time()
print(‘run time is %s‘ %(stop-start))

如果并发的多个任务是I/O密集型：多线程效率高
from multiprocessing import Process
from threading import Thread
import threading
import os,time
def work():
time.sleep(2)
print(‘===>‘)

if name == ‘main‘:
l=[]
print(os.cpu_count()) #本机为4核
start=time.time()
for i in range(400):
# p=Process(target=work) #耗时12s多,大部分时间耗费在创建进程上
p=Thread(target=work) #耗时2s多
l.append(p)
p.start()
for p in l:
p.join()
stop=time.time()
print(‘run time is %s‘ %(stop-start))

应用：多线程用于IO密集型，如socket，爬虫，web
多进程用于计算密集型，如金融分析

十一、死锁现象与递归锁
死锁： 是指两个或两个以上的进程或线程在执行过程中，因争夺资源而造成的一种互相等待的现象，若无外力作用，它们都将无法推进下去。此时称系统处于死锁状态或系统产生了死锁，这些永远在互相等待的进程称为死锁进程。

from threading import Thread,Lock
import time
mutexA=Lock()
mutexB=Lock()

class MyThread(Thread):
def run(self):
self.func1()
self.func2()
def func1(self):
mutexA.acquire()
print(‘33[41m%s 拿到A锁33[0m‘ %self.name)

    mutexB.acquire()
    print('33[42m%s 拿到B锁33[0m' %self.name)
    mutexB.release()

    mutexA.release()

def func2(self):
    mutexB.acquire()
    print('33[43m%s 拿到B锁33[0m' %self.name)
    time.sleep(2)

    mutexA.acquire()
    print('33[44m%s 拿到A锁33[0m' %self.name)
    mutexA.release()

    mutexB.release()

if name == ‘main‘:
for i in range(10):
t=MyThread()
t.start()

递归锁
递归锁，在Python中为了支持在同一线程中多次请求同一资源，python提供了可重入锁RLock。这个RLock内部维护着一个Lock和一个counter变量，counter记录了acquire的次数，从而使得资源可以被多次require。直到一个线程所有的acquire都被release，其他的线程才能获得资源。上面的例子如果使用RLock代替Lock，则不会发生死锁，二者的区别是：递归锁可以连续acquire多次，而互斥锁只能acquire一次

from threading import Thread,RLock
import time

mutexA=mutexB=RLock() #一个线程拿到锁，counter加1,该线程内又碰到加锁的情况，则counter继续加1，这期间所有其他线程都只能等待，等待该线程释放所有锁，即counter递减到0为止

class MyThread(Thread):
def run(self):
self.func1()
self.func2()
def func1(self):
mutexA.acquire()
print(‘33[41m%s 拿到A锁33[0m‘ %self.name)

    mutexB.acquire()
    print('33[42m%s 拿到B锁33[0m' %self.name)
    mutexB.release()

    mutexA.release()

def func2(self):
    mutexB.acquire()
    print('33[43m%s 拿到B锁33[0m' %self.name)
    time.sleep(2)

    mutexA.acquire()
    print('33[44m%s 拿到A锁33[0m' %self.name)
    mutexA.release()

    mutexB.release()

if name == ‘main‘:
for i in range(10):
t=MyThread()
t.start()

十二、信号量，Event，定时器
信号量
信号量也是一把锁，可以指定信号量为5，对比互斥锁同一时间只能有一个任务抢到锁去执行，信号量同一时间可以有5个任务拿到锁去执行，如果说互斥锁是合租房屋的人去抢一个厕所，那么信号量就相当于一群路人争抢公共厕所，公共厕所有多个坑位，这意味着同一时间可以有多个人上公共厕所，但公共厕所容纳的人数是一定的，这便是信号量的大小

from threading import Thread,Semaphore
import threading
import time

def func():
sm.acquire()
print(‘%s get sm‘ %threading.current_thread().getName())
time.sleep(3)
sm.release()

if name == ‘main‘:
sm=Semaphore(5)
for i in range(23):
t=Thread(target=func)
t.start()

Semaphore管理一个内置的计数器，
每当调用acquire()时内置计数器-1；
调用release() 时内置计数器+1；
计数器不能小于0；当计数器为0时，acquire()将阻塞线程直到其他线程调用release()。

Event
线程的一个关键特性是每个线程都是独立运行且状态不可预测。如果程序中的其 他线程需要通过判断某个线程的状态来确定自己下一步的操作,这时线程同步问题就会变得非常棘手。为了解决这些问题,我们需要使用threading库中的Event对象。 对象包含一个可由线程设置的信号标志,它允许线程等待某些事件的发生。在 初始情况下,Event对象中的信号标志被设置为假。如果有线程等待一个Event对象, 而这个Event对象的标志为假,那么这个线程将会被一直阻塞直至该标志为真。一个线程如果将一个Event对象的信号标志设置为真,它将唤醒所有等待这个Event对象的线程。如果一个线程等待一个已经被设置为真的Event对象,那么它将忽略这个事件, 继续执行

from threading import Event
event.isSet()：返回event的状态值；
event.wait()：如果 event.isSet()==False将阻塞线程；
event.set()： 设置event的状态值为True，所有阻塞池的线程激活进入就绪状态， 等待操作系统调度；
event.clear()：恢复event的状态值为False。

from threading import Thread,Event
import threading
import time,random
def conn_mysql():
count=1
while not event.is_set():
if count > 3:
raise TimeoutError(‘链接超时‘)
print(‘<%s>第%s次尝试链接‘ % (threading.current_thread().getName(), count))
event.wait(0.5)
count+=1
print(‘<%s>链接成功‘ %threading.current_thread().getName())

def check_mysql():
print(‘33[45m[%s]正在检查mysql33[0m‘ % threading.current_thread().getName())
time.sleep(random.randint(2,4))
event.set()
if name == ‘main‘:
event=Event()
conn1=Thread(target=conn_mysql)
conn2=Thread(target=conn_mysql)
check=Thread(target=check_mysql)

conn1.start()
conn2.start()
check.start()

定时器
定时器，指定n秒后执行某操作
from threading import Timer

def hello():
print("hello, world")

t = Timer(1, hello)
t.start() # after 1 seconds, "hello, world" will be printed

十三、线程queue
线程queue有三种不同的用法
1，class queue.Queue(maxsize=0) #队列：先进先出
import queue

q=queue.Queue()
q.put(‘first‘)
q.put(‘second‘)
q.put(‘third‘)

print(q.get())
print(q.get())
print(q.get())
2，class queue.LifoQueue(maxsize=0) #堆栈：last in fisrt out
import queue

q=queue.LifoQueue()
q.put(‘first‘)
q.put(‘second‘)
q.put(‘third‘)

print(q.get())
print(q.get())
print(q.get())

3，class queue.PriorityQueue(maxsize=0) #优先级队列：存储数据时可设置优先级的队列
import queue

q=queue.PriorityQueue()

put进入一个元组,元组的第一个元素是优先级(通常是数字,也可以是非数字之间的比较),数字越小优先级越高

q.put((20,‘a‘))
q.put((10,‘b‘))
q.put((30,‘c‘))

print(q.get())
print(q.get())
print(q.get())

十四、进程池与线程池
官网：https://docs.python.org/dev/library/concurrent.futures.html
concurrent.futures模块提供了高度封装的异步调用接口
ThreadPoolExecutor：线程池，提供异步调用
ProcessPoolExecutor: 进程池，提供异步调用
Both implement the same interface, which is defined by the abstract Executor class.

基本方法
1、submit(fn, *args, **kwargs) 异步提交任务
2、map(func, *iterables, timeout=None, chunksize=1) 取代for循环submit的操作
3、shutdown(wait=True) 相当于进程池的pool.close()+pool.join()操作
wait=True，等待池内所有任务执行完毕回收完资源后才继续
wait=False，立即返回，并不会等待池内的任务执行完毕
但不管wait参数为何值，整个程序都会等到所有任务执行完毕
submit和map必须在shutdown之前
4、result(timeout=None) 取得结果
5、add_done_callback(fn) 回调函数

进程池
用法：
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor

import os,time,random
def task(n):
print(‘%s is runing‘ %os.getpid())
time.sleep(random.randint(1,3))
return n**2

if name == ‘main‘:

executor=ProcessPoolExecutor(max_workers=3)

futures=[]
for i in range(11):
    future=executor.submit(task,i)
    futures.append(future)
executor.shutdown(True)
print('+++>')
for future in futures:
    print(future.result())

线程池
用法
把ProcessPoolExecutor换成ThreadPoolExecutor，其余用法全部相同

map方法
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor

import os,time,random
def task(n):
print(‘%s is runing‘ %os.getpid())
time.sleep(random.randint(1,3))
return n**2

if name == ‘main‘:

executor=ThreadPoolExecutor(max_workers=3)

# for i in range(11):
#     future=executor.submit(task,i)

executor.map(task,range(1,12)) #map取代了for+submit

回调函数
可以为进程池或线程池内的每个进程或线程绑定一个函数，该函数在进程或线程的任务执行完毕后自动触发，并接收任务的返回值当作参数，该函数称为回调函数

from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
from multiprocessing import Pool
import requests
import json
import os

def get_page(url):
print(‘<进程%s> get %s‘ %(os.getpid(),url))
respone=requests.get(url)
if respone.status_code == 200:
return {‘url‘:url,‘text‘:respone.text}

def parse_page(res):
res=res.result()
print(‘<进程%s> parse %s‘ %(os.getpid(),res[‘url‘]))
parse_res=‘url:<%s> size:[%s] ‘ %(res[‘url‘],len(res[‘text‘]))
with open(‘db.txt‘,‘a‘) as f:
f.write(parse_res)

if name == ‘main‘:
urls=[
‘https://www.baidu.com‘,
‘https://www.python.org‘,
‘https://www.openstack.org‘,
‘https://help.github.com/‘,
‘http://www.sina.com.cn/‘
]

p=ProcessPoolExecutor(3)
for url in urls:
    p.submit(get_page,url).add_done_callback(parse_page)

#parse_page拿到的是一个future对象obj，需要用obj.result()拿到结果