码迷,mamicode.com
首页 > 编程语言 > 详细

python并发编程(一):多线程,多进程

时间:2019-05-09 13:25:44      阅读:106      评论:0      收藏:0      [点我收藏+]

标签:ali   cal   状态   ini   设置   通过   长度   get   未来   

‘‘‘
多进程(线程)编程:
1. 进程和线程
2. 并发与并行
3. 同步和异步
4. 阻塞和非阻塞
5. 进程/线程的调度模型
‘‘‘ # 综述

‘‘‘
进程和线程:
1. 进程:
1) 是一个程序在数据集上的一次动态执行过程
2) 由程序, 数据集, 进程控制模块组成

2. 线程:
1) CPU的一个最小执行单元,
2) 线程的出现是为了降低进程间切换的消耗
3) 实现在一个进程内的并发
4) 由线程ID, 程序计数器, 寄存器集合, 堆栈组成

3. 进程和线程的关系:
1) 进程是线程的容器, 程序至少有一个进程, 一个进程至少有一个线程
2) 一个进程内的线程们共享进程资源, 线程几乎不拥有资源
3) 进程之间的资源不共享

4. 多进程/多线程的切换方式:
1) IO阻塞
2) 时间轮询
‘‘‘ # 进程和线程

‘‘‘
并发和并行:
1. 并发: 系统具有处理多个任务的能力,(可能不是并行, 如快速切换)

2. 并行: 系统具有同时处理多个任务的能力

3. 并发和并行的关系: 并行是并发的子集
‘‘‘ # 并发和并行

‘‘‘
同步和异步:
1. 同步: 当进程(线程)执行IO操作时, 没得到结果就等待

2. 异步: 当进程(线程)执行IO操作时, 不等待, 数据接收成功后, 再回来处理

3. 同步和异步时针对任务的调度方式而言(等待/不等待)
‘‘‘ # 同步和异步

‘‘‘
阻塞和非阻塞:
1. 阻塞: 当进程(线程)执行IO操作时, 该进程(线程)被挂起, 直到有结果时才被激活

2. 非阻塞: 当进程(线程)执行IO操作时, 该进程(线程)不会被挂起, 没有结果也返回

3. 阻塞和非阻塞时针对进程(线程)的调度(挂起/不挂起)
‘‘‘ # 阻塞和非阻塞

‘‘‘
进程(线程)的调度模型:
1. 进程(线程)的三种状态: 就绪, 执行, 阻塞
2. 调度模型:

------执 行--------------------
| ^_______ |
等待IO | 执行其他线程
| 执行该线程 |
v | |
阻 塞 ---有效IO--> 就 绪 <-------
‘‘‘ # 进程(线程)调度模型

‘‘‘
PythonGIL全局解释器锁
1. Python没有多线程: 同一个时刻, 在一个进程内(无论有多少线程), Python解释器只允许执行一个线程

2. 一般任务类型:
1) IO密集型 --> IO切换(阻塞即切换) 无影响
2) 计算密集型 --> 轮询切换 Python不适合用于计算密集型任务

3. 解决GIL锁的方法:
多进程 + 协程 (但主要还是为IO密集型服务的)
‘‘‘ # GIL全局锁

‘‘‘
多线程使用:
1. 导入模块 import threading
2. 开启线程的方法:
1) 直接开启:
1> 创建Thread对象
t = threading.Thread(target=func,args=(a,),kwargs={‘k1‘:v1})
2> 开启线程
t.start
2) 继承开启:
1> 继承Thread, 重写run()方法[即要执行的函数]
class MyThread(threading.Thread):
def __init__(self):
super(MyThread,self).__init__()
# 其他初始化代码,如要传给函数值
def run(self): pass
2. 创建实例,开启显线程
t = MyThread()
t.start() # 会调用其run()方法

3. 子线程和主线程的关系:
1> 主线程和子线程分别执行, 主线程会等待最后一个子线程执行完毕后, 关闭

2> 子线程t.join()到主线程, 会阻塞主线程, 执行完该子线程后主线程继续执行, 最后主线程等待最后一个子线程执行完毕后, 关闭

3> 在子线程开启之前, 申明该子线程为t.setDaemon(True)守护线程, 最后主线程不会等待该子线程, 其他线程执行完就关闭主线程

4. 线程的属性和方法:
1) 对象属性和方法: t.attr t.func()
t.func():
t.start() # 启动线程
t.join() # 阻塞当前线程
t.setDaemon(True) # 设置为守护线程
t.run() # 在当前线程下调用该线程的run方法

t.getName() # 获取线程名
t.setName() # 设置线程名

t.isAlive() # 判断线程是否被激活
t.isDaemon() # 判断线程是否为守护线程
t.attr:
t.name # 线程名
t.daemon = True/False # 是否为守护线程

2) threading方法:
threading.currentThread() # 当前线程对象
threading.enumerate() # 返回一个List包含当前所有活跃的线程对象
thread.activeCount() # 相等于len(threading.enumerate())
‘‘‘ # 多线程: 使用, Mulit_Threading.py

‘‘‘
线程之间通信:
一、同步锁(Lock):
1. 多线程操作公共变量的问题:
一个线程还未来得及对公共变量进行操作, 就又有其他线程来访问该公共变量并修改了值

2. 同步锁的作用: 部分串行
将一部分代码打包成一个整体(加锁), 同一个时刻只有获得锁的线程才能运行锁内的代码
Lock = threading.Lock()
Lock.acquire()
# code
Lock.release()

3. GIL下的Lock: 线程A,线程B,线程C,...
#1. 多线程争夺GIL, A抢到GIL的获得运行权限
#2. ALock则执行锁内代码
...
#3. 多线程继续争夺GIL, B抢到GIL
#4. 此时A尚未执行完锁内代码, 锁未释放; B没有Lock, 阻塞
...
#5. 直到A再次抢到GIL,执行完锁内代码,释放Lock
4. joinLock:
t.join()是将该线程整体阻塞到当前线程, 即整体串行
Lock是部分串行

5. 应用: 账户模型


二、死锁: 同步锁不能解决锁中锁的问题
一把锁不能重用导致的, 后面的代码需要前面的锁, 但前面的锁已经被其他线程抢到了

考虑: 线程A, 线程B, Lock_A(Lock_B), Lock_B(Lock_A)
1. 线程A先竞争到了Lock_A, 又竞争到了Lock_B, 执行锁中代码
2. 线程A执行完了, 顺序释放Lock_B, Lock_A
3. 线程A竞争到了Lock_B, 线程B竞争到了Lock_A
4. 死锁, 线程A在等Lock_A, 线程B在等Lock_B

三、递归锁(RLock): 解决死锁问题
可重入锁, 可递归使用, 没用完就不会释放
1. 原理:
1) RLock内部维护着一个Lock和一个counter变量,counter记录了acquire的次数,从而使得资源可以被多次require
2) 直到一个线程所有的acquire都被release,其他的线程才能获得资源
2.使用:
RLock = threading.RLock()
with RLock:
#code 1
with RLock:
#code 2
‘‘‘ # 多线程通信: 同步锁(threading.Lock), 死锁, 递归锁(threading.RLock)

‘‘‘
同步对象(Event):
1.作用: 多线程运行中, 每个线程都是独立运行, Event保证其他线程需要等待某个线程的指令才能执行

2. 使用: 通过event.wait()来阻塞线程,event.wait()后的代码必须等待某一个线程event.set()执行了才能执行
#1. Event = threading.Event()
#2. 指定某个线程为大哥, Event.set()发出指令
#3. 将其他线程设为小弟, Event.wait(), 要等待大哥发出指令后,Event.wait()之后代码才会执行
执行后将Event.clear() 等待下一个指令
3. 方法:
Event.isSet()
Event.set()

Event.wait()
Event.clear()
‘‘‘ # 多线程通信: 同步对象(threading.Event)

‘‘‘
信号量(Semaphore):
1. 信号量与进(线)程池的区别:
信号量是, 可以开启无限多的进程(线程), 但某一时刻只允许num个进程(线程)修改数据(涉及加锁)
进程池是, 始终只开启num个进程(没有加锁)

2. 使用:
#1. Semaphore = threading.Semaphore(num) // 创建一个num线程的Lock
#2. Semaphore.acquire() // 从池子中取一个线程, 没有则阻塞
# code
Semaphore.release() // 释放一个空间
‘‘‘ # 多线程通信: 信号量(threading.Semaphore)

‘‘‘
线程队列(queue):
1. 线程安全:
1) list不是线程安全的数据类型
2) 保证线程安全的方法:
1> 加锁
2> 线程队列
2. 队列类型:
q = queue.Queue(num) # 正常队列, FIFO
q = queue.LifoQueue(num) # , FILO
q = queue.PriorityQueue(num) # 优先队列
: q.put([num,data]) # num从小到大, 优先级从高到低
q.get() # 返回一个list

3. 线程队列保证线程安全

q.put() # 向队列中添加数据, 队列为满则会阻塞 // q.put_nowait() # 添加, 队列为满报异常
q.get() # 从队列中取数据, 队列为空则会阻塞 // q.get_nowait() # 取值, 队列为空报异常

另一种阻塞方式:
q.task_done() <--> q.join()

4. 其他方法
q.qsize() # 返回当前队列长度
q.empty() # 队列是否为空
q.full() # 队列是否为满
‘‘‘ # 多线程通信: 线程队列(queue.Queue)

‘‘‘
线程局部变量(ThreadLocal):
1. 作用: 每个线程只能访问公共变量, ThreadLocal为每个线程创建了自己的变量
2. 用法:
#1. thread_local = threading.local() // 创建全局的thread_local对象
#2. thread_local.attr = value // 在每个线程内, 线程可通过thread_local的属性设定变量
// thread_local的相同属性在各个线程中不一样, 互不影响
3. 实质:
#1. thread_local对象内部维护一个dict, key为不同线程的ID, value为这些线程存储值
#2. thread_local.attr调用属性值时, 实际上是以当前线程的IDkey, 去找值
‘‘‘ # 多线程通信: 线程局部变量(threading.local)

‘‘‘
多进程使用:
1. 导入模块 import threading
2. 开启进程的方法:
1) 直接开启:
1> 创建Process对象
p = mulitprocessing.Process(target=func,args=(a,),kwargs={‘k1‘:v1})
2> 开启线程
p.start
2) 继承开启:
1> 继承Process, 重写run()方法[即要执行的函数]
class MyProcess(multiprocessing.Process):
def __init__(self):
super(MyThread,self).__init__()
# 其他初始化代码,如要传给函数值
def run(self): pass
2. 创建实例,开启显线程
p = MyProcess()
p.start() # 会调用其run()方法

3. 子进程和主进程的关系:
1> 主进程和子进程分别执行, 主线程会等待最后一个子进程执行完毕后, 关闭

2> 子进程p.join()到主进程, 会阻塞主线程, 执行完该子进程后主进程继续执行, 最后主进程等待最后一个子进程执行完毕后, 关闭

3> 在子进程开启之前, 申明该子进程为p.daemon = True守护进程, 最后主线程不会等待该子线程, 其他进程执行完就关闭主线程

4. 进程的属性和方法:
1) 对象属性和方法: p.attr p.func()
p.func():
p.start() # 启动进程
p.join() # 阻塞当前进程
p.run() # 在当前进程下调用该进程的run方法

p.isAlive() # 判断进程是否被激活
p.terminate() # 结束进程
p.attr:
p.name # 线程名
p.daemon = True/False # 是否为守护线程
p.num/p.pid # 进程号

2) os方法:
os.getpid() # 获取当前进程号
os.getppid() # 获取当前进程父进程号
‘‘‘ # 多进程: 使用, Mulit_Process.py

‘‘‘
多进程通信:
1. 锁机制: Lock/ RLock/ Event/ Semaphore 同多线程
2. 进程队列: multiprocessing.Queue 同多线程
进程间资源不能共享, 因此在主线程中创建的Lock, RLock, Event, Semaphore, Queue, 要作为参数传入到子线程中

3. 管道(Pipe): 通过管道两端的conn双向通信
用法:
#1. conn1, conn2 = multiprocessing.Pipe() //在主进程中创建
#2. conn1, conn2作为参数传入进程中
#3. 每一个connconn.send(), conn.recv()方法
#4. conn.recv()为空阻塞, conn.send()为空不阻塞

4. 共享数据(Manager): 所有进程都可访问的数据块
用法:
#1. manager = multiprocessing.Manager() // 主进程中创建
#2. 通过manager对象可创建公共的
list, dict // 也需要传入
Lock, RLock, Event, Semaphore, Queue // 也需要传入
#3. 各个进程可访问这些数据集, 注意进程不安全的(除了Queue)要加锁

5. 进程池(Pool): 与信号量类似,但没有加锁。
信号量是,可以开无限多的进程(线程),但某一时刻只允许num个进程(线程)修改数据
进程池是,从始至终只开num个进程
‘‘‘ # 多进程通信: 锁机制, 进程队列, Pipe, Manager, Pool

python并发编程(一):多线程,多进程

标签:ali   cal   状态   ini   设置   通过   长度   get   未来   

原文地址:https://www.cnblogs.com/lancelotxly/p/10837806.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!