写在前面
最坏的结果，不过是大器晚成；
一、异常处理
- 1.语法错误导致的异常
- 这种错误，根本过不了python解释器的语法检测，必须在程序运行前就修正；
- 2.逻辑上的异常
- 即逻辑错误，例如除零错误；
- 异常相关信息：异常的追踪信息 + 异常类型 + 异常值
- 异常种类
ArithmeticError
AssertionError
AttributeError
BaseException
BufferError
BytesWarning
DeprecationWarning
EnvironmentError
EOFError
Exception
FloatingPointError
FutureWarning
GeneratorExit
ImportError
ImportWarning
IndentationError
IndexError
IOError
KeyboardInterrupt
KeyError
LookupError
MemoryError
NameError
NotImplementedError
OSError
OverflowError
PendingDeprecationWarning
ReferenceError
RuntimeError
RuntimeWarning
StandardError
StopIteration
SyntaxError
SyntaxWarning
SystemError
SystemExit
TabError
TypeError
UnboundLocalError
UnicodeDecodeError
UnicodeEncodeError
UnicodeError
UnicodeTranslateError
UnicodeWarning
UserWarning
ValueError
Warning
ZeroDivisionError
异常类别
- 异常示例：
# IndexError: list index out of range
list1 = [1,2]
list1[7] # KeyError: 'k9'
dict1 = {
'k1':'v1',
'k2':'v2',
}
dict1['k9'] # ValueError: invalid literal for int() with base 10: 'standby'
str = 'standby'
int(str) ...
- 3.异常处理
- try...except... （可以写多个except）
- 从上向下匹配，匹配到了就不再匹配下面的except，类似于 iptables
try:
# msg=input('>>:')
# int(msg) #ValueError print(x) #NameError d={'a':1}
d['b'] #KeyError l=[1,2]
l[10] #IndexError 1+'asdfsadfasdf' #TypeError except ValueError as e:
print('ValueError: %s' % e)
except NameError as e:
print('NameError: %s' % e)
except KeyError as e:
print(e) print('=============>')
- 万能异常 Exception
try:
# d={'a':1}
# d['b'] #KeyError l=[1,2]
l[10] #IndexError 1+'asdfsadfasdf' #TypeError except Exception as e:
print('捕获到异常，异常类型：%s，异常值：%s' % (type(e),e)) print('=============>')
- try...except...else...finally... （finally 里面主要是做一些清理工作）
try:
# 1+'asdfsadfasdf' #TypeError
print('aaaaaa') except Exception as e:
print('捕获到异常，异常类型：%s，异常值：%s' % (type(e), e))
else:
print('没有异常时发生会执行')
finally:
print('有没有异常都会执行')
- 4.异常扩展
- 主动抛出异常
try:
raise TypeError('类型错误')
except Exception as e:
print(e)
- 自定义异常类型
class standbyException(BaseException):
def __init__(self,msg):
self.msg=msg
def __str__(self):
return self.msg
try:
raise standbyException('--->> 自定义异常类型')
except standbyException as e:
print(e)
- 断言 assert
# 待补充...
二、操作系统
- 0.参考：操作系统简介
- 1.基础概念
- 1.精简的说的话，操作系统就是一个协调、管理和控制计算机硬件资源和软件资源的控制程序；
- 2.操作系统管理硬件，提供系统调用（接口，比方说：文件）；对资源请求进行有序调度处理；
- 3.操作系统处在用户应用程序和计算机硬件之间，本质也是一个软件
- 4.操作系统由操作系统的内核（运行于内核态，管理硬件资源）以及系统调用（运行于用户态，为应用程序员写的应用程序提供系统调用接口）两部分组成，所以，单纯的说操作系统是运行于内核态的，是不准确的。
# 1.隔离复杂度，提供简单易用的接口
隐藏了丑陋的硬件调用接口，为应用程序员提供调用硬件资源的更好，更简单，更清晰的模型（系统调用接口）。
应用程序员有了这些接口后，就不用再考虑操作硬件的细节，专心开发自己的应用程序即可。 比如，磁盘资源的抽象是文件系统（C盘，D盘，E盘...下的目录及文件），
有了文件的概念，我们直接打开文件，读或者写就可以了，
无需关心记录是否应该使用修正的调频记录方式，以及当前电机的状态等细节
# 2.将应用程序对硬件资源的竞态请求变得有序化
例如：很多应用软件其实是共享一套计算机硬件，
比方说有可能有三个应用程序同时需要申请打印机来输出内容，
那么a程序竞争到了打印机资源就打印，
然后可能是b竞争到打印机资源，也可能是c，这就导致了无序；
打印机可能打印一段a的内容然后又去打印c...，
操作系统的一个功能就是将这种无序变得有序；
- 2.相关概念
- 1.批处理
- 把一堆人的输入攒成一大波输入；把一堆人的输出攒成一大波输出；节省了机时；
- 2.多道程序设计
- 空间上的复用
- 将内存分为几部分，每个部分放入一个程序，这样，同一时间内存中就有了多道程序；
空间上的复用最大的问题是：
程序之间的内存必须分割，这种分割需要在硬件层面实现，由操作系统控制。
如果内存彼此不分割，则一个程序可以访问另外一个程序的内存， 1.首先丧失的是安全性：
比如你的qq程序可以访问操作系统的内存，这意味着你的qq可以拿到操作系统的所有权限。 2.其次丧失的是稳定性：
某个程序崩溃时有可能把别的程序的内存也给回收了，
比方说把操作系统的内存给回收了，则操作系统崩溃。
- 时间上的复用
- 快速的上下文切换
当一个资源在时间上复用时，不同的程序或用户轮流使用它，第一个程序获取该资源使用结束后，在轮到第二个。。。第三个。。。 例如：只有一个cpu，多个程序需要在该cpu上运行，操作系统先把cpu分给第一个程序；
在这个程序运行的足够长的时间（时间长短由操作系统的算法说了算）或者遇到了I/O阻塞，操作系统则把cpu分配给下一个程序；
以此类推，直到第一个程序重新被分配到了cpu然后再次运行，由于cpu的切换速度很快，给用户的感觉就是这些程序是同时运行的，或者说是并发的，或者说是伪并行的。
至于资源如何实现时间复用，或者说谁应该是下一个要运行的程序，以及一个任务需要运行多长时间，这些都是操作系统的工作。
当一个程序在等待I/O时，另一个程序可以使用cpu，如果内存中可以同时存放足够多的作业，则cpu的利用率可以接近100%；
- 3.分时操作系统
把处理机的运行时间分成很短的时间片，按时间片轮流把处理机分配给各联机作业使用； 分时操作系统，在多个程序之间切换，按时间片切换；
第三代计算机广泛采用了必须的保护硬件（程序之间的内存彼此隔离）之后，分时系统才开始流行；
操作系统的作用：
1.把硬件丑陋复杂的接口隐藏起来，给应用程序提供简单易用的接口
2.管理、调度进程，并且把进程之间对硬件的竞争变得有序化 多道技术：
1.产生背景：为了实现单CPU下的并发效果
2.分为两部分：
1.空间上的复用（内存里放入多个程序，必须实现硬件层面的隔离）
2.时间上的复用（复用的是CPU的时间片；快速切换）
什么情况下进行切换？
1.正在执行的任务遇到的阻塞（例如I/O）
2.正在执行的任务运行时间过长 进程：正在运行的一个过程/任务
由操作系统负责调度，然后由CPU负责执行 并发：伪并行，单核+多道实现
并行：只有多核才能实现真正的并行 同步：打电话
异步：发短信 进程的创建：
1.系统初始化的时候
2.与用户交互：双击一个EXE
3.在执行一个进程的过程当中，调用了Popen/os.fork
4.批处理任务 系统调用：
Linux：fork
Windows：CreateProcess Linux与Windows下的进程的区别：
1.linux下的进程有父子关系，Windows下的进程没有这个关系；
2.Linux下创建一个新的进程，需要拷贝父进程的地址空间；Windows下从最开始创建进程，两个进程之间就是不一样的； 进程的三种状态：
1.就绪
2.运行
3.阻塞 进程间快速切换，前提是在切换之前需要保存当前进程当前的状态
yield 就有这种操作系统级别的 保存状态 的功能
课上记的一些笔记，作为补充
三、进程
- 0.参考：Cpython解释器支持的进程与线程
- 1.进程的概念
- 进程：正在进行的一个过程或者说一个任务。而负责执行任务则是cpu；起源于操作系统，是操作系统最核心的概念；
- 程序仅仅只是一堆代码而已，而进程指的是程序的运行过程；
- 同一个程序执行两次，那也是两个进程，比如打开暴风影音，虽然都是同一个软件，但是一个可以播放西游记，一个可以播放天龙八部；
An executing instance of a program is called a process. Each process provides the resources needed to execute a program.
A process has a virtual address space, executable code, open handles to system objects,
a security context, a unique process identifier, environment variables,
a priority class, minimum and maximum working set sizes, and at least one thread of execution.
Each process is started with a single thread,
often called the primary thread, but can create additional threads from any of its threads. 程序并不能单独运行，只有将程序装载到内存中，系统为它分配资源才能运行，而这种执行的程序就称之为进程。
程序和进程的区别就在于：程序是指令的集合，它是进程运行的静态描述文本；进程是程序的一次执行活动，属于动态概念。 在多道编程中，我们允许多个程序同时加载到内存中，在操作系统的调度下，可以实现并发地执行。这是这样的设计，大大提高了CPU的利用率。
进程的出现让每个用户感觉到自己独享CPU，因此，进程就是为了在CPU上实现多道编程而提出的。
- 2.并发与并行
- 1.并发：是伪并行，即看起来是同时运行。单个cpu+多道技术就可以实现并发，（并行也属于并发）；
- 2.并行：同时运行，只有具备多个cpu才能实现真正意义上的并行；
- 单核下，可以利用多道技术；多个核，每个核也都可以利用多道技术（多道技术是针对单核而言的）
有四个核，六个任务，这样同一时间有四个任务被执行，假设分别被分配给了cpu1，cpu2，cpu3，cpu4； 一旦任务1遇到I/O就被迫中断执行，此时任务5就拿到cpu1的时间片去执行，这就是单核下的多道技术； 而一旦任务1的I/O结束了，操作系统会重新调用它；
(需知进程的调度、分配给哪个cpu运行，由操作系统说了算)，可能被分配给四个cpu中的任意一个去执行
多道技术：
内存中同时存入多道（多个）程序，cpu从一个进程快速切换到另外一个，
使每个进程各自运行几十或几百毫秒； 这样，虽然在某一个瞬间，一个cpu只能执行一个任务，
但在1秒内，cpu却可以运行多个进程，这就给人产生了并行的错觉，即伪并发，
以此来区分多处理器操作系统的真正硬件并行（多个cpu共享同一个物理内存）
- 3.同步和异步
同步就是指一个进程在执行某个请求的时候，若该请求需要一段时间才能返回信息，
那么这个进程将会一直等待下去，直到收到返回信息才继续执行下去； 异步是指进程不需要一直等下去，而是继续执行下面的操作，不管其他进程的状态。
当有消息返回时系统会通知进程进行处理，这样可以提高执行的效率。 同步：打电话
异步：发短息、MySQL主从复制
- 4.multiprocess简单介绍
python中的多线程无法利用多核优势;
如果想要充分地使用多核CPU的资源（os.cpu_count()查看），在python中大部分情况需要使用多进程。 Python提供了非常好用的多进程包multiprocessing。 1.multiprocessing模块用来开启子进程，并在子进程中执行我们定制的任务（比如函数），该模块与多线程模块threading的编程接口类似。 2.multiprocessing模块的功能众多：支持子进程、通信和共享数据、执行不同形式的同步，提供了Process、Queue、Pipe、Lock、Pool等组件。 需要再次强调的一点是：
与线程不同，进程没有任何共享状态，进程修改的数据，改动仅限于该进程内;
- 5.Process介绍
- 1.参数介绍
Process([group [, target [, name [, args [, kwargs]]]]])
由该类实例化得到的对象，表示一个子进程中的任务（尚未启动） 强调：
1. 需要使用关键字的方式来指定参数
2. args指定的为传给target函数的位置参数，是一个元组形式，必须有逗号 group 参数未使用，值始终为None
target 表示调用对象，即子进程要执行的任务
args 表示调用对象的位置参数元组，args=(1,2,'egon',)
kwargs 表示调用对象的字典,kwargs={'name':'egon','age':18}
name 为子进程的名称
- 2.方法介绍
p.start()
启动进程，并调用该子进程中的p.run() p.run()
进程启动时运行的方法，正是它去调用target指定的函数，
我们自定义类的类中一定要实现该方法 p.terminate()
强制终止进程p，不会进行任何清理操作，
如果p创建了子进程，该子进程就成了僵尸进程，使用该方法需要特别小心这种情况。
如果p还保存了一个锁那么也将不会被释放，进而导致死锁 p.is_alive()
如果p仍然运行，返回True p.join([timeout]) hold住的是主进程
主线程等待p终止（强调：是主线程处于等的状态，而p是处于运行的状态）。
timeout是可选的超时时间，需要强调的是，p.join只能join住start开启的进程，
而不能join住run开启的进程
- 3.属性介绍
p.daemon
默认值为False，如果设为True，代表p为后台运行的守护进程，当p的父进程终止时，p也随之终止；
并且设定为True后，p不能创建自己的新进程，必须在p.start()之前设置； p.name:进程的名称 p.pid：进程的pid p.exitcode
进程在运行时为None、如果为–N，表示被信号N结束(了解即可) p.authkey
进程的身份验证键,默认是由os.urandom()随机生成的32字符的字符串。
这个键的用途是为涉及网络连接的底层进程间通信提供安全性，
这类连接只有在具有相同的身份验证键时才能成功（了解即可）
- 4.注意
注意：在windows中Process()必须放到# if __name__ == '__main__':下 Since Windows has no fork,
the multiprocessing module starts a new Python process and imports the calling module.
If Process() gets called upon import,
then this sets off an infinite succession of new processes (or until your machine runs out of resources).
This is the reason for hiding calls to Process() inside if __name__ == "__main__"
since statements inside this if-statement will not get called upon import. 由于Windows没有fork，多处理模块启动一个新的Python进程并导入调用模块。
如果在导入时调用Process()，那么这将启动无限继承的新进程（或直到机器耗尽资源）。
这是隐藏对Process()内部调用的原因，
使用if __name__ == “__main __”，这个if语句中的语句将不会在导入时被调用。
- 6.进程的创建
1. 在UNIX中该系统调用是：fork；
fork会创建一个与父进程一模一样的副本，
二者有相同的存储映像、同样的环境字符串和同样的打开文件；
（在shell解释器进程中，执行一个命令就会创建一个子进程） 2. 在windows中该系统调用是：CreateProcess，
CreateProcess既处理进程的创建，也负责把正确的程序装入新进程。
关于创建的子进程，UNIX和windows 1.相同的是：进程创建后，父进程和子进程有各自不同的地址空间（多道技术要求物理层面实现进程之间内存的隔离），
任何一个进程的在其地址空间中的修改都不会影响到另外一个进程； 2.不同的是：在UNIX中，子进程的初始地址空间是父进程的一个副本，
提示：子进程和父进程是可以有只读的共享内存区的。
但是对于windows系统来说，从一开始父进程与子进程的地址空间就是不同的。
- 创建进程示例1：
# Windows上调用Process，可执行代码一定要放到 __main__ 里
from multiprocessing import Process
import time,random def func(name):
print('%s is running...' % name)
# time.sleep(random.randint(1,3))
time.sleep(1)
print('%s run end.' % name) if __name__ == '__main__':
# p1 = Process(target=func,args=('standby',))
p1 = Process(target=func,args=('standby',),name='sub-P1') # 指定进程的名字
p1.start()
print(p1.name)
print('Parent running')
time.sleep(1)
print('Parent run end') ---
sub-P1
Parent running
standby is running...
Parent run end
standby run end.
- 创建进程示例2：使用继承的方式，必须在类中定义一个run()方法
# 继承Process类
from multiprocessing import Process
import time, random class Foo(Process):
def __init__(self, name):
super().__init__() # 调用父类Process的init方法进行初始化
self.name = name
def run(self): # 必须定义run()方法
print('%s is running...' % self.name)
# time.sleep(random.randint(1,3))
time.sleep(1)
print('%s run end.' % self.name)
if __name__ == '__main__':
p1 = Foo('standby')
p1.start() # start() 会自动调用 run()
print('Parent running...')
time.sleep(1)
print('Parent run end.') ---
Parent running...
standby is running...
Parent run end.
standby run end.
- 7.进程的状态
就绪状态：进程已经准备好，已分配到所需资源，只要分配到CPU就能够立即运行；
如果进程运行时间片使用完也会进入就绪状态； 执行状态：进程处于就绪状态被调度后，进程进入执行状态； 阻塞状态：正在执行的进程由于某些事件（I/O请求，申请缓存区失败）而暂时无法运行，进程受到阻塞；
在满足请求时进入就绪状态等待系统调用；
其实在两种情况下会导致一个进程在逻辑上不能运行：
1. 进程挂起是自身原因，遇到I/O阻塞，便要让出CPU让其他进程去执行，这样保证CPU一直在工作
2. 与进程无关，是操作系统层面，可能会因为一个进程占用时间过多，或者优先级等原因，而调用其他的进程去使用CPU。
- 8.Process对象的常用方法和属性
- 1.daemon=True与join()
p.daemon=True 之后，如果在子进程p内再创建子进程就会报错：
'daemonic processes are not allowed to have children'
# 示例1：没有join也不设置daemon
from multiprocessing import Process
def say(name):
print('%s say hello.' % name) if __name__ == '__main__':
p = Process(target=say,args=('standby',))
p.start() # 创建新的进程需要时间，所以主线程先打印；但子进程也会打印，因为主线程需要等待子进程执行完毕，才结束，避免出现僵尸进程（没有父进程的子进程）
print('这是主线程') ---结果---
这是主线程
standby say hello. ===================>
# 示例2：设置daemon=True，没有join
from multiprocessing import Process
def say(name):
print('%s say hello.' % name) if __name__ == '__main__':
p = Process(target=say,args=('standby',))
p.daemon = True #一定要在p.start()前设置；设置p为守护进程，禁止p创建子进程；并且父进程结束，p跟着一起结束
p.start()
print('这是主线程') ---结果---
这是主线程 ===================>
# 示例3：有join，没有设置daemon=True
from multiprocessing import Process
def say(name):
print('%s say hello.' % name) if __name__ == '__main__':
p = Process(target=say,args=('standby',))
p.start()
p.join() # 阻塞了主线程，使得子进程先执行完毕，主线程才继续往下执行
print('这是主线程') ---结果---
standby say hello.
这是主线程 ===================>
# 示例4：设置daemon=True，有join
from multiprocessing import Process
def say(name):
print('%s say hello.' % name) if __name__ == '__main__':
p = Process(target=say,args=('standby',))
p.daemon = True #把p设置成守护进程
p.start()
p.join()
print('这是主线程') ---结果---
standby say hello.
这是主线程 ===================>
===================>
===================>
# 示例5：子进程为什么会不打印？
from multiprocessing import Process
import time
def say(name):
print('%s say hello.' % name)
time.sleep(2)
print('%s say bye.' % name) if __name__ == '__main__':
p = Process(target=say,args=('standby',))
p.daemon = True #把p设置成守护进程
p.start()
p.join(0.001) # 主线程等待p的结束，等0.0001秒就不再等了；
print('这是主线程') ---结果---
这是主线程 ===================>
# 示例6：把主线程阻塞的时间改大一点，改为0.1s
from multiprocessing import Process
import time
def say(name):
print('%s say hello.' % name)
time.sleep(2)
print('%s say bye.' % name) if __name__ == '__main__':
p = Process(target=say,args=('standby',))
p.daemon = True #把p设置成守护进程
p.start()
p.join(0.1)
print('这是主线程') ---结果---
standby say hello.
这是主线程 ===================>
# 示例7：把主线程阻塞的时间改大一点，改为2s
from multiprocessing import Process
import time
def say(name):
print('%s say hello.' % name)
time.sleep(2)
print('%s say bye.' % name) if __name__ == '__main__':
p = Process(target=say,args=('standby',))
p.daemon = True #把p设置成守护进程
p.start()
p.join(2)
print('这是主线程') ---结果---
standby say hello.
这是主线程 ===================>
# 示例8：把主线程阻塞的时间改大一点，改为2.1s
from multiprocessing import Process
import time
def say(name):
print('%s say hello.' % name)
time.sleep(2)
print('%s say bye.' % name) if __name__ == '__main__':
p = Process(target=say,args=('standby',))
p.daemon = True #把p设置成守护进程
p.start()
p.join(2.1)
print('这是主线程') ---结果---
standby say hello.
standby say bye.
这是主线程
- 2.terminate() 与 is_alive()
# 示例1：查看terminate之后的状态
from multiprocessing import Process
def say(name):
print('%s say hello.' % name)
print('%s say bye.' % name) if __name__ == '__main__':
p = Process(target=say,args=('standby',))
print(p.is_alive())
p.start()
p.terminate()
print(p.is_alive())
print('这是主线程')
print(p.is_alive()) ---结果---
False
True
这是主线程
True ========================>
# 示例2：在主线程中sleep一下
from multiprocessing import Process
import time
def say(name):
print('%s say hello.' % name)
print('%s say bye.' % name) if __name__ == '__main__':
p = Process(target=say,args=('standby',))
print(p.is_alive())
p.start()
p.terminate() # 关闭进程,不会立即关闭,所以is_alive立刻查看的结果可能还是存活；
print(p.is_alive())
print('这是主线程')
time.sleep(0.01)
print(p.is_alive()) ---结果---
False
True
这是主线程
False
- 3.name 和 pid
from multiprocessing import Process
import os
def say():
print('Say hello，子进程id：%s' % os.getpid()) if __name__ == '__main__':
p = Process(target=say)
p.start()
print('子进程的名字是：%s，子进程id：%s' % (p.name,p.pid))
print('这是主线程，主线程id：%s' % os.getpid()) ---结果---
子进程的名字是：Process-1，子进程id：1612
这是主线程，主线程id：6004
Say hello，子进程id：1612
四、线程
- 为什么要有线程？
有了进程为什么还要线程？ 进程有很多优点，它提供了多道编程，
让我们感觉我们每个人都拥有自己的CPU和其他资源，可以提高计算机的利用率。
很多人就不理解了，既然进程这么优秀，为什么还要线程呢？
其实，仔细观察就会发现进程还是有很多缺陷的，主要体现在两点上： 1.进程只能在一个时间干一件事，如果想同时干两件事或多件事，进程就无能为力了。 2.进程在执行的过程中如果阻塞，例如等待输入，整个进程就会挂起；
即使进程中有些工作不依赖于输入的数据，也将无法执行。
- 线程示例：
进程与进程之间的资源是隔离的；
一个进程里的多个线程共享进程的资源； 例子：编译一个文档，有三个功能，接收用户输入+格式化+定期保存
1.用三个进程实现；但进程间数据是隔离的，这样就需要维护三份资源数据；
2.用1个进程挂三个线程实现，三个线程共享一份资源；
- 线程是什么？
- 进程只是用来把资源集中到一起（进程只是一个资源单位，或者说资源集合），而线程才是cpu上的执行单位；
- 线程是操作系统能够进行运算调度的最小单位。它被包含在进程之中，是进程中的实际运作单位；
- 一条线程指的是进程中一个单一顺序的控制流，一个进程中可以并发多个线程，每条线程并行执行不同的任务；
- 进程之间是竞争关系，线程之间是协作关系；
A thread is an execution context,
which is all the information a CPU needs to execute a stream of instructions. Suppose you're reading a book, and you want to take a break right now,
but you want to be able to come back and resume reading from the exact point where you stopped.
One way to achieve that is by jotting down the page number, line number, and word number.
So your execution context for reading a book is these 3 numbers. If you have a roommate, and she's using the same technique,
she can take the book while you're not using it, and resume reading from where she stopped.
Then you can take it back, and resume it from where you were. Threads work in the same way.
A CPU is giving you the illusion that it's doing multiple computations at the same time.
It does that by spending a bit of time on each computation.
It can do that because it has an execution context for each computation.
Just like you can share a book with your friend, many tasks can share a CPU. On a more technical level, an execution context (therefore a thread) consists of the values of the CPU's registers. Last: threads are different from processes.
A thread is a context of execution,
while a process is a bunch of resources associated with a computation.
A process can have one or many threads. Clarification:
the resources associated with a process include memory pages (all the threads in a process have the same view of the memory),
file descriptors (e.g., open sockets),
and security credentials (e.g., the ID of the user who started the process).
- 线程和进程的区别
# 1
Threads share the address space of the process that created it;
Processes have their own address space. # 2
Threads have direct access to the data segment of its process;
Processes have their own copy of the data segment of the parent process. # 3
Threads can directly communicate with other threads of its process;
Processes must use interprocess communication to communicate with sibling processes. # 4
New threads are easily created;
New processes require duplication of the parent process. # 5
Threads can exercise considerable control over threads of the same process; Processes can only exercise control over child processes. # 6
Changes to the main thread (cancellation, priority change, etc.) may affect the behavior of the other threads of the process;
Changes to the parent process does not affect child processes.
- 创建进程的开销要远大于创建线程的开销；
# 创建 500 个线程
import time
from threading import Thread
def work():
a = 99999
b = 101001010010101010
str1 = 'axaxxchaxchnahxalx'
str2 = 'axaxxcedw2312haxchnahxalx'
str3 = '121212axaxxchaxchnahxalx'
dic = {'k1':'v1','k2':'v2'} if __name__ == '__main__':
start_time = time.time()
t_l = []
for i in range(500):
t=Thread(target=work)
t_l.append(t)
t.start()
for t in t_l:
t.join()
stop_time = time.time()
print('Run time is %s' % (stop_time-start_time))
# Run time is 0.05900001525878906 # ++++++++++++++++++++++++++++++++++ # 创建 500 个进程
import time
from multiprocessing import Process
def work():
a = 99999
b = 101001010010101010
str1 = 'axaxxchaxchnahxalx'
str2 = 'axaxxcedw2312haxchnahxalx'
str3 = '121212axaxxchaxchnahxalx'
dic = {'k1':'v1','k2':'v2'} if __name__ == '__main__':
start_time = time.time()
p_l = []
for i in range(500):
p=Process(target=work)
p_l.append(p)
p.start()
for p in p_l:
p.join()
stop_time = time.time()
print('Run time is %s' % (stop_time-start_time))
# Run time is 19.552000045776367
关于线程和协程更多参见： ...
五、多线程和多进程
- 多进程
- 模拟创建多个子进程，未阻塞的情况
from multiprocessing import Process
import time,random def func(name):
print('%s is running...' % name)
time.sleep(random.randint(1,3))
print('%s run end.' % name) if __name__ == '__main__':
# p1 = Process(target=func,args=('standby',))
p1 = Process(target=func,args=('进程1',),name='sub-P1')
p2 = Process(target=func,args=('进程2',),name='sub-P2')
p3 = Process(target=func,args=('进程3',),name='sub-P3')
p4 = Process(target=func,args=('进程4',),name='sub-P4')
sub_p_lits = [p1,p2,p3,p4]
for p in sub_p_lits:
p.start()
print('Parent running')
time.sleep(1)
print('Parent run end') ---
Parent running
进程2 is running...
进程4 is running...
进程3 is running...
进程1 is running...
Parent run end
进程3 run end.
进程2 run end.
进程4 run end.
进程1 run end.
- 模拟创建多个子进程，join阻塞的情况：
from multiprocessing import Process
import time,random def func(name):
print('%s is running...' % name)
time.sleep(random.randint(1,3))
print('%s run end.' % name) if __name__ == '__main__':
# p1 = Process(target=func,args=('standby',))
p1 = Process(target=func,args=('进程1',),name='sub-P1')
p2 = Process(target=func,args=('进程2',),name='sub-P2')
p3 = Process(target=func,args=('进程3',),name='sub-P3')
p4 = Process(target=func,args=('进程4',),name='sub-P4')
sub_p_lits = [p1,p2,p3,p4]
for p in sub_p_lits:
p.start()
for p in sub_p_lits:
p.join()
print('Parent running')
time.sleep(1)
print('Parent run end') ---
进程2 is running...
进程1 is running...
进程3 is running...
进程4 is running...
进程2 run end.
进程3 run end.
进程4 run end.
进程1 run end.
Parent running
Parent run end
- socket + Process 实现并发处理多个客户端连接
#!/usr/bin/python
# -*- coding:utf-8 -*-
# Server端 from multiprocessing import Process
from socket import *
server=socket(AF_INET,SOCK_STREAM)
server.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
server.bind(('127.0.0.1',8080))
server.listen(5) def talk(conn,addr):
while True: #通讯循环
try:
msg=conn.recv(1024)
if not msg:break
conn.send(msg.upper())
except Exception:
break
if __name__ == '__main__':
while True: #链接循环
conn,addr=server.accept()
print(addr)
p=Process(target=talk,args=(conn,addr))
p.start()
#!/usr/bin/python
# -*- coding:utf-8 -*-
# Client端 from socket import *
client=socket(AF_INET,SOCK_STREAM)
client.connect(('127.0.0.1',8080)) while True:
msg=input('>>: ').strip()
if not msg:continue
client.send(msg.encode('utf-8'))
msg=client.recv(1024)
print(msg.decode('utf-8'))
# 多进程实现socket存在的问题：
每来一个客户端，都在服务端开启一个进程；
如果并发来一个万个客户端，要开启一万个进程吗？
你自己尝试着在你自己的机器上开启一万个，10万个进程试一试。（导致机器死机） 解决方法：进程池
五、进程池
- 进程池有啥用？
Pool可以提供指定数量的进程，供用户调用；
当有新的请求提交到pool中时，如果池还没有满，那么就会创建一个新的进程用来执行该请求；
但如果池中的进程数已经达到规定最大值，那么该请求就会等待，直到池中有进程结束，就重用进程池中的进程。 注意：是重用进程，而不是新建，Pool维护指定个数的进程，来循环执行很多的任务，进程的id都是不变的！
- Pool
Pool([numprocess [,initializer [, initargs]]]):创建进程池 numprocess:要创建的进程数，如果省略，将默认使用cpu_count()的值；
- 常用方法
参考：http://www.cnblogs.com/congbo/archive/2012/08/23/2652490.html
p.apply(func [, args [, kwargs]])
进程池中的工作进程执行func(*args,**kwargs)，然后返回结果； 是同步的； p.apply_async(func [, args [, kwargs]])
与apply用法一致，但它是非阻塞的且支持结果返回后进行回调； 是异步的；
主进程循环运行过程中不等待apply_async的返回结果；
在主进程结束后，即使子进程还未返回整个程序也会退出。
虽然 apply_async是非阻塞的，但其返回结果的get方法却是阻塞的：
如使用result.get()会阻塞主进程。 p.close()
关闭进程池，即不再像进程池中提交任务。如果所有操作持续挂起，它们将在工作进程终止前完成; P.join()
主进程阻塞等待子进程的退出， join方法要在close或terminate之后使用; p.terminate()
结束工作进程，不再处理未处理的任务;
- apply()，同步
1.apply本质上就是apply_async().get() 2.apply_async().get()会返回结果，但同时也会阻塞，导致变为串行；
- apply()示例
from multiprocessing import Pool
import time
def Foo(i):
time.sleep(1)
return i + 100
if __name__ == '__main__':
start_time = time.time()
pool = Pool(5)
res_l = []
for i in range(5):
res = pool.apply(func=Foo, args=(i,))
res_l.append(res)
pool.close()
for res in res_l:
print(res)
print('---> end')
stop_time = time.time()
print('Run time is: %s' % (stop_time-start_time)) ---结果---
100
101
102
103
104
---> end
Run time is: 5.174000024795532
- apply_async()，异步非阻塞
如果使用异步提交的任务，主进程需要使用join，等待进程池内任务都处理完，然后可以用get收集结果，
否则，主进程结束，进程池可能还没来得及执行，也就跟着一起结束了；
调用join之前，一定要先调用close() 函数，否则会出错； close()执行后不会有新的进程加入到pool,join函数等待素有子进程结束；
- apply_async() 示例1：没有后面的join() 和 get()，则程序整体结束，进程池中的任务还没来得及全部执行完也都跟着主进程一起结束了；
# 在主线程内没有使用join()的情况
from multiprocessing import Pool
import time
def Foo(i):
time.sleep(1)
return i + 100
def Bar(arg):
print('--->exec done:', arg)
if __name__ == '__main__':
start_time = time.time()
pool = Pool(5)
for i in range(5):
pool.apply_async(func=Foo, args=(i,), callback=Bar)
print('end')
pool.close()
stop_time = time.time()
print('Run time is: %s' % (stop_time-start_time)) ---结果---
end
Run time is: 0.08100008964538574
- apply_async() 示例2：主线程里没写join()，但是使用get()获取进程池中进程执行的结果
from multiprocessing import Pool
import time
def Foo(i):
time.sleep(1)
return i + 100
def Bar(arg):
print('--->exec done:', arg)
if __name__ == '__main__':
start_time = time.time()
pool = Pool(5)
res_l = []
for i in range(5):
res = pool.apply_async(func=Foo, args=(i,))
res_l.append(res)
pool.close() # 关闭进程池，不再向进程池中提交任务；
for res in res_l:
print(res.get()) # 使用get()获取进程池中进程的执行结果
print('end')
stop_time = time.time()
print('Run time is: %s' % (stop_time-start_time)) ---结果---
100
101
102
103
104
end
Run time is: 1.2239999771118164
- apply_async() 示例3：使用join()但不使用get()
# 在主线程内使用 join() 的情况
from multiprocessing import Pool
import time
def Foo(i):
time.sleep(1)
return i + 100
def Bar(arg):
print('--->exec done:', arg)
if __name__ == '__main__':
start_time = time.time()
pool = Pool(5)
for i in range(5):
pool.apply_async(func=Foo, args=(i,), callback=Bar)
pool.close() # 关闭进程池，不再向进程池中提交任务；
pool.join() # 进程池中进程执行完毕后再关闭，如果注释那么程序直接关闭；
print('end')
stop_time = time.time()
print('Run time is: %s' % (stop_time-start_time)) ---结果---
--->exec done: 100
--->exec done: 101
--->exec done: 102
--->exec done: 103
--->exec done: 104
end
Run time is: 1.3329999446868896
- 进程池Pool改写socket并发通信，避免使用多进程的缺陷问题
# 进程池 Pool实现socket服务端 from multiprocessing import Pool
from socket import *
import os
server=socket(AF_INET,SOCK_STREAM)
server.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
server.bind(('127.0.0.1',8090))
server.listen(5) def talk(conn,addr):
print('子进程id：%s' % os.getpid())
while True: #通讯循环
try:
msg=conn.recv(1024)
if not msg:break
conn.send(msg.upper())
except Exception:
break
if __name__ == '__main__':
print("cpu_count: %s" % os.cpu_count())
pool=Pool()
while True: #链接循环
conn,addr=server.accept()
print(addr)
# pool.apply(talk,args=(conn,addr))
pool.apply_async(talk,args=(conn,addr))
# socket客户端 from socket import *
client=socket(AF_INET,SOCK_STREAM)
client.connect(('127.0.0.1',8090)) while True:
msg=input('>>: ').strip()
if not msg:continue
client.send(msg.encode('utf-8'))
msg=client.recv(1024)
print(msg.decode('utf-8'))
cpu_count: 4 <--- Pool()默认取这个值
('127.0.0.1', 4944)
子进程id：2872 <---进程池中第一个进程的id
('127.0.0.1', 4945)
子进程id：1076 <---进程池中第二个进程的id
('127.0.0.1', 4948)
子进程id：5544 <---进程池中第三个进程的id
('127.0.0.1', 4951)
子进程id：5500 <---进程池中第四个进程的id
('127.0.0.1', 4952)
('127.0.0.1', 4953)
子进程id：2872 <=== 后面新来的连接复用原来的进程连接
子进程id：1076 <=== 如果进程池中4个进程都在用，则后面新来的连接将处在阻塞的状态，一旦有进程释放，新连接就会复用被释放的进程id
('127.0.0.1', 4975)
子进程id：5544
('127.0.0.1', 4982)
- 回调函数应用
回调函数 是主进程在处理； 谁有返回值就通知主进程，然后主进程去执行回调函数里的操作；
- 不需要回调函数的场景：如果在主进程中等待进程池中所有任务都执行完毕后，再统一处理结果，则无需回调函数；此种情况可使用get()获取执行结果； - 需要回调函数的场景：进程池中任何一个任务一旦处理完了，就立即告知主进程：我好了额，你可以处理我的结果了。 主进程则调用一个函数去处理该结果，该函数即回调函数；
我们可以把耗时间（阻塞）的任务放到进程池中，然后指定回调函数（主进程负责执行），
这样主进程在执行回调函数时就省去了I/O的过程，直接拿到的是任务的结果。
- 应用示例：爬虫
from multiprocessing import Pool
import requests
import re
import json def get_page(url,pattern): # 判断并下载网页，并返回给回调函数
response=requests.get(url)
if response.status_code == 200:
return (response.text,pattern)
else:
print('response.status_code not 200.')
def parse_page(info): # 作为回调函数，按照写好的正则去解析网页内容
page_content,pattern=info
res=re.findall(pattern,page_content)
for item in res:
dic={
'index':item[0],
'title':item[1],
'actor':item[2].strip()[3:],
'time':item[3][5:],
'score':item[4]+item[5]
}
with open('maoyan.txt','a',encoding='utf-8') as f: # 按照定义好的字典，写入到文件中
f.write('%s\n' % json.dumps(dic))
if __name__ == '__main__':
pattern1=re.compile(r'<dd>.*?board-index.*?>(\d+)<.*?title="(.*?)".*?star.*?>(.*?)<.*?releasetime.*?>(.*?)<.*?integer.*?>(.*?)<.*?fraction.*?>(.*?)<',re.S)
url_dic={
'http://maoyan.com/board':pattern1,
}
p=Pool()
for url,pattern in url_dic.items():
p.apply_async(get_page,args=(url,pattern),callback=parse_page)
p.close()
p.join() # 等待进程池的进程任务都执行完毕
with open('maoyan.txt', mode='r', encoding='utf-8') as rf:
item_list = rf.readlines()
for item in item_list:
res = json.loads(item)
print(res) # 读取文件内容，查看爬取的内容 ---maoyan.txt内容---
{"actor": "\u738b\u9719,\u4fdd\u5251\u950b,\u8463\u52c7", "index": "1", "score": "9.3", "time": "2017-06-30", "title": "\u8840\u6218\u6e58\u6c5f"}
{"actor": "\u674e\u5fae\u6f2a,\u4ea6\u98ce", "index": "2", "score": "9.3", "time": "2017-06-16", "title": "\u91cd\u8fd4\u00b7\u72fc\u7fa4"}
{"actor": "\u9ad8\u5f3a,\u4e8e\u6708\u4ed9,\u674e\u7389\u5cf0", "index": "3", "score": "9.2", "time": "2017-06-09", "title": "\u5fe0\u7231\u65e0\u8a00"}
{"actor": "\u6768\u57f9,\u5c3c\u739b\u624e\u5806,\u65af\u6717\u5353\u560e", "index": "4", "score": "8.9", "time": "2017-06-20", "title": "\u5188\u4ec1\u6ce2\u9f50"}
{"actor": "\u6234\u592b\u00b7\u5e15\u7279\u5c14,\u9c81\u59ae\u00b7\u739b\u62c9,\u5927\u536b\u00b7\u6587\u7ff0", "index": "5", "score": "8.8", "time": "2017-06-22", "title": "\u96c4\u72ee"}
{"actor": "\u76d6\u5c14\u00b7\u52a0\u6735,\u514b\u91cc\u65af\u00b7\u6d3e\u6069,\u7f57\u5bbe\u00b7\u6000\u7279", "index": "6", "score": "8.6", "time": "2017-06-02", "title": "\u795e\u5947\u5973\u4fa0"}
{"actor": "\u8521\u5353\u598d,\u5468\u67cf\u8c6a,\u949f\u6b23\u6f7c", "index": "7", "score": "8.5", "time": "2017-06-23", "title": "\u539f\u8c05\u4ed677\u6b21"}
{"actor": "\u738b\u6653\u5f64,\u674e\u6654,\u6d2a\u6d77\u5929", "index": "8", "score": "8.1", "time": "2017-05-27", "title": "\u4e09\u53ea\u5c0f\u732a2"}
{"actor": "\u590f\u96e8,\u95eb\u59ae,\u6f58\u658c\u9f99", "index": "9", "score": "8.0", "time": "2017-06-29", "title": "\u53cd\u8f6c\u4eba\u751f"}
{"actor": "\u6768\u5e42,\u970d\u5efa\u534e,\u91d1\u58eb\u6770", "index": "10", "score": "7.9", "time": "2017-06-29", "title": "\u9006\u65f6\u8425\u6551"} ---打印结果---
{'actor': '王霙,保剑锋,董勇', 'index': '1', 'score': '9.3', 'time': '2017-06-30', 'title': '血战湘江'}
{'actor': '李微漪,亦风', 'index': '2', 'score': '9.3', 'time': '2017-06-16', 'title': '重返·狼群'}
{'actor': '高强,于月仙,李玉峰', 'index': '3', 'score': '9.2', 'time': '2017-06-09', 'title': '忠爱无言'}
{'actor': '杨培,尼玛扎堆,斯朗卓嘎', 'index': '4', 'score': '8.9', 'time': '2017-06-20', 'title': '冈仁波齐'}
{'actor': '戴夫·帕特尔,鲁妮·玛拉,大卫·文翰', 'index': '5', 'score': '8.8', 'time': '2017-06-22', 'title': '雄狮'}
{'actor': '盖尔·加朵,克里斯·派恩,罗宾·怀特', 'index': '6', 'score': '8.6', 'time': '2017-06-02', 'title': '神奇女侠'}
{'actor': '蔡卓妍,周柏豪,钟欣潼', 'index': '7', 'score': '8.5', 'time': '2017-06-23', 'title': '原谅他77次'}
{'actor': '王晓彤,李晔,洪海天', 'index': '8', 'score': '8.1', 'time': '2017-05-27', 'title': '三只小猪2'}
{'actor': '夏雨,闫妮,潘斌龙', 'index': '9', 'score': '8.0', 'time': '2017-06-29', 'title': '反转人生'}
{'actor': '杨幂,霍建华,金士杰', 'index': '10', 'score': '7.9', 'time': '2017-06-29', 'title': '逆时营救'}
六、进程同步（锁）
进程之间数据不共享，但是共享同一套文件系统；
所以访问同一个文件，或同一个打印终端，是没有问题的；
1.共享同一打印终端
多个进程同时执行打印操作：
发现会有多行内容打印到一行的现象（多个进程共享并抢占同一个打印终端，乱了）
- 示例：
# 共享同一个打印终端
import os,time
from multiprocessing import Process
class Logger(Process):
def __init__(self):
super().__init__()
# super(Logger,self).__init__()
def run(self):
# time.sleep(1)
print(self.name,'pid: %s' % os.getpid()) if __name__ == '__main__':
for i in range(100): # i5, 4核，起1W个进程就会死机，慎跑...
l=Logger()
l.start()
2.共享同一个文件
共享同一个文件;
有的同学会想到，既然可以用文件共享数据，那么进程间通信用文件作为数据传输介质就可以了啊; 可以，但是有问题：
1.效率
2.需要自己加锁处理
- 示例：
#多进程共享一套文件系统
from multiprocessing import Process
def write_to_file(file,mode,num):
with open(file,mode=mode,encoding='utf-8') as wf:
wf.write(num)
if __name__ == '__main__':
for i in range(50):
p = Process(target=write_to_file,args=('a.txt','a',str(i)))
p.start()
3.锁（互斥锁）
加锁的目的是为了保证多个进程修改同一块数据时，同一时间只能有一个修改，即串行的修改； 没错，速度是慢了，牺牲了速度而保证了数据安全； 进程之间数据隔离，但是共享一套文件系统，
因而可以通过文件来实现进程直接的通信，但问题是必须自己加锁处理
- 模拟抢票程序
#文件db的内容为：{"count":1}
#注意一定要用双引号，不然json无法识别；
- 没有加锁的抢票程序：由于没有加锁，所以存在多个进程同时去读写文件的情况，导致一张票被多个人抢到；
from multiprocessing import Process,Lock
import json
import time
import random
def work(dbfile,name,lock):
with open(dbfile,encoding='utf-8') as f:
dic=json.loads(f.read())
if dic['count'] > 0:
dic['count'] -= 1
time.sleep(random.randint(1,3)) # 模拟网络延迟
with open(dbfile,'w',encoding='utf-8') as f:
f.write(json.dumps(dic))
print('\033[43m%s 抢票成功\033[0m' %name)
else:
print('\033[45m%s 抢票失败\033[0m' %name) if __name__ == '__main__':
lock=Lock() # 进程间不共享数据，所以需要把lock当做参数传进入
p_l=[]
for i in range(100):
p=Process(target=work,args=('a.txt','用户%s' %i,lock))
p_l.append(p)
p.start()
for p in p_l:
p.join()
print('主进程') ---结果---
用户6 抢票成功
用户19 抢票失败
用户0 抢票失败
用户3 抢票成功
用户45 抢票成功
用户38 抢票失败
用户10 抢票成功
用户51 抢票失败
用户5 抢票失败
用户58 抢票失败
用户8 抢票失败
用户54 抢票失败
用户87 抢票失败
用户71 抢票失败
用户94 抢票失败
用户66 抢票失败
用户93 抢票失败
用户1 抢票成功
用户61 抢票失败
用户31 抢票失败
用户70 抢票失败
用户13 抢票失败
用户77 抢票失败
用户92 抢票失败
用户7 抢票失败
用户34 抢票失败
用户44 抢票失败
用户23 抢票失败
用户29 抢票失败
用户33 抢票失败
用户41 抢票失败
用户82 抢票失败
用户86 抢票失败
用户39 抢票失败
用户43 抢票失败
用户90 抢票失败
用户17 抢票失败
用户28 抢票失败
用户14 抢票失败
用户67 抢票失败
用户48 抢票失败
用户37 抢票失败
用户24 抢票失败
用户63 抢票失败
用户46 抢票失败
用户25 抢票失败
用户74 抢票失败
用户47 抢票失败
用户80 抢票失败
用户57 抢票失败
用户11 抢票失败
用户30 抢票失败
用户96 抢票失败
用户73 抢票失败
用户91 抢票失败
用户22 抢票失败
用户20 抢票失败
用户89 抢票失败
用户83 抢票失败
用户98 抢票失败
用户53 抢票失败
用户88 抢票失败
用户79 抢票失败
用户78 抢票失败
用户49 抢票失败
用户64 抢票失败
用户95 抢票失败
用户18 抢票失败
用户97 抢票失败
用户59 抢票失败
用户72 抢票失败
用户42 抢票失败
用户21 抢票失败
用户32 抢票失败
用户4 抢票失败
用户27 抢票失败
用户65 抢票失败
用户62 抢票失败
用户99 抢票失败
用户55 抢票失败
用户81 抢票失败
用户15 抢票失败
用户40 抢票失败
用户69 抢票失败
用户85 抢票失败
用户16 抢票失败
用户50 抢票失败
用户26 抢票失败
用户60 抢票失败
用户75 抢票失败
用户35 抢票失败
用户68 抢票失败
用户36 抢票失败
用户52 抢票失败
用户84 抢票失败
用户76 抢票失败
用户12 抢票成功
用户2 抢票成功
用户9 抢票成功
用户56 抢票成功
主进程
- 加锁的情况：牺牲执行速度，保护了数据安全性，只有一个用户能抢到票；
from multiprocessing import Process,Lock
import json
import time
import random
def work(dbfile,name,lock):
# lock.acquire()
with lock:
with open(dbfile,encoding='utf-8') as f:
dic=json.loads(f.read())
if dic['count'] > 0:
dic['count']-=1
time.sleep(random.randint(1,3)) #模拟网络延迟
with open(dbfile,'w',encoding='utf-8') as f:
f.write(json.dumps(dic))
print('\033[43m%s 抢票成功\033[0m' %name)
else:
print('\033[45m%s 抢票失败\033[0m' %name)
# lock.release()
if __name__ == '__main__':
start_time = time.time()
lock=Lock() # 进程间不共享数据，所以需要把lock当做参数传进入
p_l=[]
for i in range(100):
p=Process(target=work,args=('a.txt','用户%s' % i,lock))
p_l.append(p)
p.start()
for p in p_l:
p.join()
stop_time = time.time()
print('Run time: %s' % (stop_time-start_time)) ---结果---
用户19 抢票成功
用户1 抢票失败
用户0 抢票失败
用户3 抢票失败
用户7 抢票失败
用户13 抢票失败
用户5 抢票失败
用户55 抢票失败
用户51 抢票失败
用户43 抢票失败
用户39 抢票失败
用户59 抢票失败
用户63 抢票失败
用户62 抢票失败
用户2 抢票失败
用户50 抢票失败
用户47 抢票失败
用户23 抢票失败
用户14 抢票失败
用户9 抢票失败
用户18 抢票失败
用户75 抢票失败
用户21 抢票失败
用户27 抢票失败
用户54 抢票失败
用户11 抢票失败
用户61 抢票失败
用户15 抢票失败
用户31 抢票失败
用户38 抢票失败
用户25 抢票失败
用户35 抢票失败
用户6 抢票失败
用户30 抢票失败
用户34 抢票失败
用户42 抢票失败
用户36 抢票失败
用户67 抢票失败
用户26 抢票失败
用户46 抢票失败
用户17 抢票失败
用户49 抢票失败
用户71 抢票失败
用户22 抢票失败
用户45 抢票失败
用户10 抢票失败
用户53 抢票失败
用户65 抢票失败
用户29 抢票失败
用户69 抢票失败
用户73 抢票失败
用户33 抢票失败
用户52 抢票失败
用户58 抢票失败
用户37 抢票失败
用户41 抢票失败
用户24 抢票失败
用户40 抢票失败
用户48 抢票失败
用户81 抢票失败
用户57 抢票失败
用户32 抢票失败
用户83 抢票失败
用户79 抢票失败
用户77 抢票失败
用户20 抢票失败
用户95 抢票失败
用户89 抢票失败
用户93 抢票失败
用户91 抢票失败
用户44 抢票失败
用户99 抢票失败
用户97 抢票失败
用户90 抢票失败
用户87 抢票失败
用户85 抢票失败
用户74 抢票失败
用户66 抢票失败
用户16 抢票失败
用户86 抢票失败
用户82 抢票失败
用户98 抢票失败
用户70 抢票失败
用户78 抢票失败
用户72 抢票失败
用户84 抢票失败
用户28 抢票失败
用户94 抢票失败
用户4 抢票失败
用户60 抢票失败
用户8 抢票失败
用户96 抢票失败
用户80 抢票失败
用户92 抢票失败
用户76 抢票失败
用户88 抢票失败
用户12 抢票失败
用户64 抢票失败
用户56 抢票失败
用户68 抢票失败
Run time: 4.325000047683716
七、进程间通信
1.操作同一个文件
- 需对文件加锁：保护共享数据；
- 见上面抢票的例子；
2.IPC，InterProcess Communication
进程彼此之间互相隔离，要实现进程间通信（IPC）， multiprocessing模块支持两种形式：队列和管道，这两种方式都是使用消息传递的；
- 1.队列（推荐使用），队列Queue是管道+锁实现的；先进先出；
# Queue类
Queue([maxsize]):创建共享的进程队列，Queue是多进程安全的队列，可以使用Queue实现多进程之间的数据传递； # 属性
maxsize是队列中允许最大项数，省略则无大小限制； # 常用方法
q.put()
用以插入数据到队列中，put方法还有两个可选参数：blocked和timeout。
如果blocked为True（默认值），并且timeout为正值，该方法会阻塞timeout指定的时间，直到该队列有剩余的空间。
如果超时，会抛出Queue.Full异常。
如果blocked为False，但该Queue已满，会立即抛出Queue.Full异常。 q.get()
可以从队列读取并且删除一个元素。get方法也有两个可选参数：blocked和timeout。
如果blocked为True（默认值），并且timeout为正值，那么在等待时间内没有取到任何元素，会抛出Queue.Empty异常。
如果blocked为False，有两种情况存在：如果Queue有一个值可用，则立即返回该值，否则，如果队列为空，则立即抛出Queue.Empty异常. q.get_nowait(): 同 q.get(False) q.put_nowait(): 同 q.put(False) q.empty()
调用此方法时q为空则返回True，该结果不可靠，比如在返回True的过程中，如果队列中又加入了项目； q.full()：调用此方法时q已满则返回True，该结果不可靠，比如在返回True的过程中，如果队列中的项目被取走； q.qsize():返回队列中目前项目的正确数量，结果也不可靠，理由同q.empty()和q.full()一样；
def put(self, obj, block=True, timeout=None): # 队列满的时候，q.put()会阻塞，直到队列不满
pass def get(self, block=True, timeout=None): # 队列空的时候，q,get()会阻塞，直到队列有新元素
pass
- 示例1
from multiprocessing import Process,Queue q = Queue(3)
q.put(1)
q.put('liu')
q.put(('a','b','c',))
print(q.full())
print(q.qsize())
print(q.get())
print(q.get())
print(q.get())
print(q.empty()) ---
True
3
1
liu
('a', 'b', 'c')
True
- 示例2：简单的生产者消费者模型
from multiprocessing import Queue, Process
import random, os, time def getter(name, queue):
print('Son process name: %s, pid: %s, ppid: %s' % (name, os.getpid(), os.getppid()))
while True:
try:
value = queue.get(block=True, timeout=3)
# block为True,就是如果队列中无数据了。
# |—————— 若timeout默认是None，那么会一直等待下去。
# |—————— 若timeout设置了时间，那么会等待timeout秒后才会抛出Queue.Empty异常
# block 为False，如果队列中无数据，就抛出Queue.Empty异常
print("Process getter get: %f" % value)
except Exception as e:
print('getter捕捉到异常：%s' % e)
break def putter(name, queue):
print('Son process name: %s, pid: %s, ppid: %s' % (name, os.getpid(), os.getppid()))
for i in range(0, 10):
time.sleep(1)
value = random.randint(1,10)
queue.put(value)
# 放入数据 put(obj[, block[, timeout]])
# 若block为True，如队列是满的：
# |—————— 若timeout是默认None，那么就会一直等下去
# |—————— 若timeout设置了等待时间，那么会等待timeout秒后，如果还是满的，那么就抛出Queue.Full.
# 若block是False，如果队列满了，直接抛出Queue.Full
print("Process putter put: %f" % value) if __name__ == '__main__':
queue = Queue()
getter_process = Process(target=getter, args=("Getter", queue))
putter_process = Process(target=putter, args=("Putter", queue))
getter_process.start()
putter_process.start()
getter_process.join()
putter_process.join()
print('Main process, pid: %s' % os.getpid()) ---结果---
Son process name: Getter, pid: 3088, ppid: 5656
Son process name: Putter, pid: 2712, ppid: 5656
Process putter put: 8.000000
Process getter get: 8.000000
Process putter put: 6.000000
Process getter get: 6.000000
Process putter put: 6.000000
Process getter get: 6.000000
Process putter put: 8.000000
Process getter get: 8.000000
Process putter put: 3.000000
Process getter get: 3.000000
Process putter put: 6.000000
Process getter get: 6.000000
Process putter put: 1.000000
Process getter get: 1.000000
Process putter put: 7.000000
Process getter get: 7.000000
Process putter put: 8.000000
Process getter get: 8.000000
Process putter put: 6.000000
Process getter get: 6.000000
getter捕捉到异常：
Main process, pid: 5656
- 2.管道（不推荐）
tail -f access.log |grep '404'
# Pipe函数
def Pipe(duplex=True):
return Connection(), Connection() Returns a pair (conn1, conn2) of Connection objects representing the ends of a pipe.
在进程之间创建一条管道，并返回元组（conn1,conn2），其中conn1，conn2表示管道两端的连接对象； 强调一点：必须在产生Process对象之前产生管道； # 参数
dumplex:默认管道是全双工的，如果将duplex射成False，conn1只能用于接收，conn2只能用于发送。
新建一个Pipe(duplex)的时候，如果duplex为True，那么创建的管道是双向的；如果duplex为False，那么创建的管道是单向的。 # 常用方法
conn1.recv():接收conn2.send(obj)发送的对象。如果没有消息可接收，recv方法会一直阻塞。如果连接的另外一端已经关闭，那么recv方法会抛出EOFError; conn1.send(obj):通过连接发送对象。obj是与序列化兼容的任意对象
- 管道Pipe建立的过程（Pipe的读写效率要高于Queue。）
进程间的Pipe基于fork机制建立。 当主进程创建Pipe的时候，Pipe的两个Connections连接的的都是主进程。 当主进程创建子进程后，Connections也被拷贝了一份。此时有了4个Connections。 此后，关闭主进程的一个Out Connection，关闭一个子进程的一个In Connection。那么就建立好了一个输入在主进程，输出在子进程的管道。
- 示例1：基于管道实现的进程间通信
from multiprocessing import Pipe, Process def son_process(pipe):
_out_pipe, _in_pipe = pipe
# 关闭fork过来的输入端
_in_pipe.close()
while True:
try:
msg = _out_pipe.recv()
print('子进程通过管道获得消息：%s' % msg)
except Exception as e:
print('子进程中捕获到异常：%s' % e)
# 当out_pipe接受不到输出的时候且输入被关闭的时候，会抛出EORFError，可以捕获并且退出子进程
break if __name__ == '__main__':
out_pipe, in_pipe = Pipe(True)
son_p = Process(target=son_process, args=((out_pipe, in_pipe),))
son_p.start()
# 等pipe被fork 后，关闭主进程的输出端
# 这样，创建的Pipe一端连接着主进程的输入，一端连接着子进程的输出口
out_pipe.close()
for x in range(20):
in_pipe.send(x)
in_pipe.close()
son_p.join()
print("主进程也结束了") ---结果---
子进程通过管道获得消息：0
子进程通过管道获得消息：1
子进程通过管道获得消息：2
子进程通过管道获得消息：3
子进程通过管道获得消息：4
子进程通过管道获得消息：5
子进程通过管道获得消息：6
子进程通过管道获得消息：7
子进程通过管道获得消息：8
子进程通过管道获得消息：9
子进程通过管道获得消息：10
子进程通过管道获得消息：11
子进程通过管道获得消息：12
子进程通过管道获得消息：13
子进程通过管道获得消息：14
子进程通过管道获得消息：15
子进程通过管道获得消息：16
子进程通过管道获得消息：17
子进程通过管道获得消息：18
子进程通过管道获得消息：19
子进程中捕获到异常：
主进程也结束了
- 示例2：基于管道实现的进程间通信
from multiprocessing import Process,Pipe
import time
def consumer(p,name):
left,right=p
left.close()
while True:
try:
baozi=right.recv()
print('%s 收到包子:%s' %(name,baozi))
except EOFError:
right.close()
break
def producer(seq,p):
left,right=p
right.close()
for i in seq:
left.send(i)
time.sleep(1)
else:
left.close()
if __name__ == '__main__':
left,right=Pipe()
c1=Process(target=consumer,args=((left,right),'c1'))
c1.start()
seq=(i for i in range(10))
producer(seq,(left,right))
right.close()
left.close()
c1.join()
print('主进程') ---结果---
c1 收到包子:0
c1 收到包子:1
c1 收到包子:2
c1 收到包子:3
c1 收到包子:4
c1 收到包子:5
c1 收到包子:6
c1 收到包子:7
c1 收到包子:8
c1 收到包子:9
主进程
- 示例3：基于管道的双向通信
from multiprocessing import Process,Pipe def adder(p,name):
server,client=p
client.close()
while True:
try:
x,y=server.recv()
print('%s 通过管道收到了：%s 和 %s' % (name,x,y))
except EOFError:
server.close()
break
res = x + y
server.send(res)
print('%s 回应消息：%s' % (name, res))
if __name__ == '__main__':
server,client=Pipe()
s1=Process(target=adder,args=((server,client),'s1'))
s1.start()
server.close()
client.send((10,20))
print(client.recv())
client.close()
s1.join()
print('主进程') ---结果---
s1 通过管道收到了：10 和 20
30
s1 回应消息：30
主进程
- 3.扩展：生产者消费者模型
- 1.主进程作为生产者，一个子进程作为消费者
# 生产者消费者模型
from multiprocessing import Process,Queue
import time
import random def consumer(q,name):
while True:
time.sleep(random.randint(1,3))
try:
res=q.get(block=True,timeout=5)
print('\033[41m消费者%s拿到了%s\033[0m' %(name,res))
except Exception as e:
print('消费者捕获异常：%s' % e)
break
def producer(seq,q,name):
for item in seq:
time.sleep(random.randint(1,3))
q.put(item)
print('\033[45m生产者%s生产了%s\033[0m' %(name,item)) if __name__ == '__main__':
q=Queue()
c=Process(target=consumer,args=(q,'standby'),)
c.start()
seq=['包子%s' % i for i in range(10)] # 列表生成式
producer(seq,q,'厨师')
print('生产者已经生产完毕') ---结果---
生产者厨师生产了包子0
生产者厨师生产了包子1
消费者standby拿到了包子0
生产者厨师生产了包子2
消费者standby拿到了包子1
消费者standby拿到了包子2
生产者厨师生产了包子3
生产者厨师生产了包子4
消费者standby拿到了包子3
生产者厨师生产了包子5
消费者standby拿到了包子4
生产者厨师生产了包子6
消费者standby拿到了包子5
生产者厨师生产了包子7
消费者standby拿到了包子6
消费者standby拿到了包子7
生产者厨师生产了包子8
消费者standby拿到了包子8
生产者厨师生产了包子9
生产者已经生产完毕
消费者standby拿到了包子9
消费者捕获异常：
- 2.开两个子进程，一个作为生产者，另一个作为消费者
from multiprocessing import Process,Queue
import time
import random def consumer(q,name):
while True:
time.sleep(random.randint(1,3))
res=q.get()
if res is None:break
print('\033[41m消费者%s拿到了%s\033[0m' %(name,res))
def producer(seq,q,name):
for item in seq:
time.sleep(random.randint(1,3))
q.put(item)
print('\033[45m生产者%s生产了%s\033[0m' %(name,item))
q.put(None) if __name__ == '__main__':
q=Queue()
c=Process(target=consumer,args=(q,'standby'),)
c.start()
seq=['包子%s' %i for i in range(10)]
p=Process(target=producer,args=(seq,q,'厨师'))
p.start()
c.join()
print('主进程') ---结果---
生产者厨师生产了包子0
消费者standby拿到了包子0
生产者厨师生产了包子1
消费者standby拿到了包子1
生产者厨师生产了包子2
生产者厨师生产了包子3
消费者standby拿到了包子2
生产者厨师生产了包子4
消费者standby拿到了包子3
生产者厨师生产了包子5
消费者standby拿到了包子4
生产者厨师生产了包子6
消费者standby拿到了包子5
生产者厨师生产了包子7
消费者standby拿到了包子6
生产者厨师生产了包子8
消费者standby拿到了包子7
生产者厨师生产了包子9
消费者standby拿到了包子8
消费者standby拿到了包子9
主进程
- 3.JoinableQueue
创建队列的另外一个类：JoinableQueue([maxsize])；
这就像是一个Queue对象，但队列允许项目的使用者通知生成者项目已经被成功处理。
通知进程是使用共享的信号和条件变量来实现的。 #参数介绍：
maxsize是队列中允许最大项数，省略则无大小限制； #方法介绍：
JoinableQueue的实例p除了与Queue对象相同的方法之外还具有： q.task_done()：
使用者使用此方法发出信号，表示q.get()的返回项目已经被处理。如果调用此方法的次数大于从队列中删除项目的数量，将引发ValueError异常；
q.join():
生产者调用此方法进行阻塞，直到队列中所有的项目均被处理。阻塞将持续到队列中的每个项目均调用q.task_done（）方法为止；
- JoinableQueue示例
from multiprocessing import Process,JoinableQueue
import time
import random def consumer(q,name):
while True:
time.sleep(random.randint(1,3))
res=q.get()
q.task_done()
print('\033[41m消费者%s拿到了%s\033[0m' %(name,res))
def producer(seq,q,name):
for item in seq:
time.sleep(random.randint(1,3))
q.put(item)
print('\033[45m生产者%s生产了%s\033[0m' %(name,item))
print('生产者生产完毕')
q.join()
print('消费者都消费完了') if __name__ == '__main__':
q=JoinableQueue()
c=Process(target=consumer,args=(q,'standby'),)
c.daemon=True #设置守护进程，主进程结束c就结束
c.start()
seq=['包子%s' %i for i in range(10)]
p=Process(target=producer,args=(seq,q,'厨师'))
p.start() p.join() #主进程等待p结束，p等待c把数据都取完，c一旦取完数据,p.join就是不再阻塞,进
# 而主进程结束,主进程结束会回收守护进程c,而且c此时也没有存在的必要了
print('主进程') ---结果---
生产者厨师生产了包子0
消费者standby拿到了包子0
生产者厨师生产了包子1
消费者standby拿到了包子1
生产者厨师生产了包子2
消费者standby拿到了包子2
生产者厨师生产了包子3
消费者standby拿到了包子3
生产者厨师生产了包子4
消费者standby拿到了包子4
生产者厨师生产了包子5
消费者standby拿到了包子5
生产者厨师生产了包子6
消费者standby拿到了包子6
生产者厨师生产了包子7
消费者standby拿到了包子7
生产者厨师生产了包子8
消费者standby拿到了包子8
生产者厨师生产了包子9
生产者生产完毕
消费者standby拿到了包子9
消费者都消费完了
主进程
3.Manager共享内存
- 示例1：不加锁的情况
# Manager实现共享内存，没加锁的例子
# 这种情况会出现同时有多个进程在写同一个内存中所共享的数据，导致最后数据不对；
from multiprocessing import Manager,Process
def func(dic):
dic['count'] -= 1 if __name__ == '__main__':
m = Manager()
dic = m.dict({'count':100})
obj_l = []
for i in range(100):
p = Process(target=func,args=(dic,))
p.start()
obj_l.append(p)
for p in obj_l:
p.join()
print(dic) ---结果---
{'count': 2}
- 示例2：第一种加锁的写法
# Manager实现共享内存，加锁实现的例子
# 第一种写法：
from multiprocessing import Manager,Process,Lock
def func(dic,lock):
lock.acquire()
dic['count'] -= 1
lock.release() if __name__ == '__main__':
lock = Lock()
m = Manager()
dic = m.dict({'count':100})
obj_l = []
for i in range(100):
p = Process(target=func,args=(dic,lock))
p.start()
obj_l.append(p)
for p in obj_l:
p.join()
print(dic) ---结果---
{'count': 0}
- 示例3.第二种加锁的写法
from multiprocessing import Manager,Process,Lock
def func(dic,lock):
with lock:
dic['count'] -= 1 if __name__ == '__main__':
lock = Lock()
# m = Manager()
with Manager() as m:
dic = m.dict({'count':100})
obj_l = []
for i in range(100):
p = Process(target=func,args=(dic,lock))
p.start()
obj_l.append(p)
for p in obj_l:
p.join()
print(dic) ---结果---
{'count': 0}
八、paramiko模块
Paramiko is a Python (2.6+, 3.3+) implementation of the SSHv2 protocol [1],
providing both client and server functionality.
While it leverages a Python C extension for low level cryptography (Cryptography),
Paramiko itself is a pure Python interface around SSH networking concepts.
- paramiko是用python语言写的一个模块，遵循SSH2协议，支持以加密和认证的方式，进行远程服务器的连接。
__all__ = [
'Transport',
'SSHClient',
'MissingHostKeyPolicy',
'AutoAddPolicy',
'RejectPolicy',
'WarningPolicy',
'SecurityOptions',
'SubsystemHandler',
'Channel',
'PKey',
'RSAKey',
'DSSKey',
'Message',
'SSHException',
'AuthenticationException',
'PasswordRequiredException',
'BadAuthenticationType',
'ChannelException',
'BadHostKeyException',
'ProxyCommand',
'ProxyCommandFailure',
'SFTP',
'SFTPFile',
'SFTPHandle',
'SFTPClient',
'SFTPServer',
'SFTPError',
'SFTPAttributes',
'SFTPServerInterface',
'ServerInterface',
'BufferedFile',
'Agent',
'AgentKey',
'HostKeys',
'SSHConfig',
'util',
'io_sleep',
]
Paramiko提供的方法
- 实例1：密码验证登录主机，远程执行命令
import paramiko
IP = '10.0.0.9'
PORT = 22
USER = 'root'
PASSWORD = '123456.'
ssh_conn = paramiko.SSHClient()
ssh_conn.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
ssh_conn.connect(IP,PORT,USER,PASSWORD,timeout=3)
except Exception as e:
print('连接失败：%s' % e)
while True:
cmd = input('Input cmd, q/Q to exit.>>>\t').strip()
if 'Q' == cmd.upper():
print('Bye...')
break
stdin,stdout,stderr = ssh_conn.exec_command(cmd)
# print(stdin.read())
res = stderr.read().decode('utf-8')
if not res:
res = stdout.read().decode('utf-8')
print(res)
ssh_conn.close()
- 实例2：密码验证登录主机，执行ftp上传下载操作
# 下载到本地
import paramiko t = paramiko.Transport(('10.0.0.9',22))
t.connect(username='root',password='123456.')
sftp = paramiko.SFTPClient.from_transport(t)
sftp.get(r'/data/pass.txt','1.txt')
t.close() # 上传到远端服务器
import paramiko t = paramiko.Transport(('10.0.0.9',22))
t.connect(username='root',password='123456.')
sftp = paramiko.SFTPClient.from_transport(t)
sftp.put(r'D:\soft\work\Python_17\day09\paramiko_demo.py','/data/paramiko_demo.py')
# sftp.get(r'/data/pass.txt','1.txt')
t.close()
- 实例3：秘钥验证登录远程主机，执行命令
参见：Pool多进程示例
九、Python GIL
参考：http://www.dabeaz.com/python/UnderstandingGIL.pdf
In CPython,
the global interpreter lock, or GIL, is a mutex that prevents multiple native threads from executing Python bytecodes at once. This lock is necessary mainly because CPython’s memory management is not thread-safe.
(However, since the GIL exists, other features have grown to depend on the guarantees that it enforces.)
- GIL并不是Python的特性，Python完全可以不依赖于GIL；
GIL使得同一时刻统一进程中只有一个线程被执行； 进程可以利用多核，但是开销大，而python的多线程开销小，但却无法利用多核优势；
十、扩展
- 1.列表生成式
seq=['数字：%s' %i for i in range(10)]
print(type(seq))
print(seq) ---结果---
<class 'list'>
['数字：0', '数字：1', '数字：2', '数字：3', '数字：4', '数字：5', '数字：6', '数字：7', '数字：8', '数字：9']
- 2.os.cpu_count()
十一、练习
要求：
题目:简单主机批量管理工具
需求:
- 主机分组
主机信息配置文件用configparser解析
可批量执行命令、发送文件，结果实时返回，执行格式如下
- batch_run -h h1,h2,h3 -g web_clusters,db_servers -cmd "df -h"
batch_scp -h h1,h2,h3 -g web_clusters,db_servers -action put -local test.py -remote /tmp/
主机用户名密码、端口可以不同
执行远程命令使用paramiko模块
批量命令需使用multiprocessing并发
代码实现：
# Config for configparser test. ;[DEFAULT]
;Admin = 'liulixin@hitedu.org'
;Description = 'Traffic Server' [QLS_GROUP]
Master = True
Slave1 = True
Slave2 = False
Slave3 = False [Master]
ip = 10.0.0.9
port = 22
user = root
password = 123456.
enable = True [Slave1]
ip = 10.0.0.8
port = 55336
user = root
password = slave1.
enable = True [Slave2]
ip = 10.0.0.7
port = 3333
user = root
password = slave2.
enable = False [Slave3]
ip = 10.0.0.6
port = 3307
user = root
password = slave3.
enable = True
my.cnf配置文件
程序代码：
#!/usr/bin/python
# -*- coding:utf-8 -*- import os
import paramiko
import configparser
from multiprocessing import Pool
file_path = os.path.abspath(__file__)
config = configparser.ConfigParser()
config.read('my.cnf',encoding='utf-8') info = '''
Command Syntax like this:
批量执行命令：batch_run -h h1,h2 -g web_clusters,db_servers -cmd "df -h"
批量拷贝文件：batch_scp -h h1,h2 -g web_clusters,db_servers -action put test.py /tmp/
退出：q/Q
''' def ssh_cmd(ip,port,user,password,cmd):
ssh_conn = paramiko.SSHClient()
ssh_conn.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
ssh_conn.connect(ip,int(port),user,password,timeout=3)
except Exception as e:
print('连接失败：%s' % e)
stdin, stdout, stderr = ssh_conn.exec_command(cmd)
ret = []
ret.append(ip)
res = stderr.read().decode('utf-8')
if not res:
res = stdout.read().decode('utf-8')
ret.append(res)
return ret
def ssh_cmd_callback(ret):
print("This is callback func of %s" % ret[0]) def scp_put(ip,port,user,password,file_to_put,remote_dir):
try:
trans = paramiko.Transport((ip,int(port)))
trans.connect(username=user, password=password)
sftp = paramiko.SFTPClient.from_transport(trans)
sftp.put(localpath=os.path.join(os.path.dirname(file_path),file_to_put),\
remotepath=r'%s/%s' % (remote_dir,file_to_put))
trans.close()
except Exception as e:
print('scp_put err: %s' % e) def scp_get(ip,port,user,password,file_to_get, local_file_name):
try:
trans = paramiko.Transport((ip, int(port)))
trans.connect(username=user, password=password)
sftp = paramiko.SFTPClient.from_transport(trans)
print(file_to_get,local_file_name)
sftp.get(file_to_get,local_file_name)
trans.close()
except Exception as e:
print('scp_get err: %s' % e) def output(res_list):
for res in res_list:
# print(res.get())
print(res.get()[1]) def get_target_ip_dict(cmd_list):
target_ip_dict = {}
if '-h' in cmd_list:
sectionos_list = config.sections()
sectionos_list.remove('QLS_GROUP')
h_index = cmd_list.index('-h')
input_host_str = cmd_list[h_index + 1]
input_host_list = input_host_str.split(',')
for host in input_host_list:
if host.capitalize() in sectionos_list:
enable = config.get(host.capitalize(), 'enable')
if 'False' == enable:
print('The %s is offline now, continue...' % host.capitalize())
continue
item_dict = {
'ip': None,
'port': None,
'user': None,
'password': None,
}
item_dict['ip'] = config.get(host.capitalize(), 'ip')
item_dict['port'] = config.get(host.capitalize(), 'port')
item_dict['user'] = config.get(host.capitalize(), 'user')
item_dict['password'] = config.get(host.capitalize(), 'password')
if host.capitalize() not in target_ip_dict:
target_ip_dict[host.capitalize()] = item_dict
else:
print('No server: %s exist.' % host)
if '-g' in cmd_list:
sectionos_list = config.sections()
g_index = cmd_list.index('-g')
input_group_str = cmd_list[g_index + 1]
input_group_list = input_group_str.split(',')
for group in input_group_list:
if group.upper() not in sectionos_list:
print('No group: %s exist.' % group)
else:
available_tag_list = []
for tag, value in config.items(group.upper()):
if 'True' == value:
available_tag_list.append(tag.capitalize())
for tag in available_tag_list:
if tag.capitalize() not in target_ip_dict:
item_dict = {
'ip': None,
'port': None,
'user': None,
'password': None,
}
item_dict['ip'] = config.get(tag.capitalize(), 'ip')
item_dict['port'] = config.get(tag.capitalize(), 'port')
item_dict['user'] = config.get(tag.capitalize(), 'user')
item_dict['password'] = config.get(tag.capitalize(), 'password')
target_ip_dict[tag.capitalize()] = item_dict
return target_ip_dict def batch_run(cmd_list):
target_ip_dict = get_target_ip_dict(cmd_list)
cmd_index = cmd_list.index('-cmd')
cmd_to_exec = ' '.join(cmd_list[cmd_index+1:])
my_pool = Pool(len(target_ip_dict))
res_list = []
for host in target_ip_dict:
res = my_pool.apply_async(func=ssh_cmd,args=(target_ip_dict[host]['ip'],\
target_ip_dict[host]['port'],\
target_ip_dict[host]['user'],\
target_ip_dict[host]['password'],\
cmd_to_exec.strip('"')),
callback=ssh_cmd_callback)
res_list.append(res)
my_pool.close()
my_pool.join()
output(res_list) def batch_scp(cmd_list):
target_ip_dict = get_target_ip_dict(cmd_list)
action_index = cmd_list.index('-action')
if 'PUT' != cmd_list[action_index+1].upper() and 'GET' != cmd_list[action_index+1].upper():
print("Scp option invaild, just support put and get option.")
return
# -action put test.py /tmp/
if 'PUT' == cmd_list[action_index+1].upper():
file_to_put = cmd_list[action_index+2]
remote_dir = cmd_list[-1]
if os.path.exists(file_to_put):
print('Start to put %s' % file_to_put)
my_pool = Pool(len(target_ip_dict))
res_list = []
for host in target_ip_dict:
res = my_pool.apply_async(func=scp_put, args=(target_ip_dict[host]['ip'],\
target_ip_dict[host]['port'],\
target_ip_dict[host]['user'],\
target_ip_dict[host]['password'],\
file_to_put,remote_dir))
res_list.append(res)
my_pool.close()
my_pool.join()
# output(res_list)
print('End to put %s' % file_to_put)
else:
print('%s not exist.' % file_to_put)
return
# -action get /path/test.py local_file_name
if 'GET' == cmd_list[action_index + 1].upper():
file_to_get = cmd_list[action_index + 2]
local_file_name = cmd_list[-1]
print('Start to get %s' % file_to_get)
my_pool = Pool(len(target_ip_dict))
res_list = []
for host in target_ip_dict:
res = my_pool.apply_async(func=scp_get, args=(target_ip_dict[host]['ip'],\
target_ip_dict[host]['port'],\
target_ip_dict[host]['user'],\
target_ip_dict[host]['password'],\
file_to_get, local_file_name))
res_list.append(res)
my_pool.close()
my_pool.join()
print('End to get %s' % file_to_get) def bye():
print('Bye...')
exit(0)
cmd_option = {
'batch_run':batch_run,
'batch_scp':batch_scp,
}
if __name__ == '__main__':
while True:
print(info)
cmd_input = input('请输入命令>>>\t').strip()
if 'Q' == cmd_input.upper():
bye()
cmd_list = cmd_input.split()
if cmd_list[0] not in cmd_option:
print('输入无效')
elif '-h' != cmd_list[1] and '-g' != cmd_list[1]:
print(type(cmd_list[1]),cmd_list[1])
print('目标主机/主机组无效')
elif '-cmd' not in cmd_list and '-action' not in cmd_list:
print('输入的操作命令不符合语法规则')
else:
cmd_option[cmd_list[0]](cmd_list)
