并发编程之多线程

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

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

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

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

4、threading模块介绍
multiprocess模块模仿了threading模块的接口；使用层面上很是类似

5、开启线程的两种方式
方式一：
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("主线程开始运行....")

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

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

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

在主进程下开启线程
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,由于同一进程内的线程之间共享进程内的数据

8、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
'''

9、守护线程
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)

10、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的存在，同一时刻同一进程中只有一个线程被执行
一、对计算来讲，cpu越多越好，可是对于I/O来讲，再多的cpu也没用（CPU密集型代码）
二、固然对运行一个程序来讲，随着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
多进程用于计算密集型，如金融分析

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

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('\033[41m%s 拿到A锁\033[0m' %self.name)

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

    mutexA.release()

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

    mutexA.acquire()
    print('\033[44m%s 拿到A锁\033[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('\033[41m%s 拿到A锁\033[0m' %self.name)

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

    mutexA.release()

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

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

    mutexB.release()

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

12、信号量，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('\033[45m[%s]正在检查mysql\033[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

十3、线程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())

十4、进程池与线程池
官网：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.

基本方法
一、submit(fn, *args, **kwargs) 异步提交任务
二、map(func, *iterables, timeout=None, chunksize=1) 取代for循环submit的操做
三、shutdown(wait=True) 至关于进程池的pool.close()+pool.join()操做
wait=True，等待池内全部任务执行完毕回收完资源后才继续
wait=False，当即返回，并不会等待池内的任务执行完毕
但无论wait参数为什么值，整个程序都会等到全部任务执行完毕
submit和map必须在shutdown以前
四、result(timeout=None) 取得结果
五、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]\n' %(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()拿到结果