第九篇 异常处理和并发编程1

一、异常处理

#1 什么是异常：程序运行时发生错误的信号（一旦出错，就会产生一个异常，
# 如果该异常没有被应用程序处理，那么该异常才会抛出来，程序也随之终止）
print('=====>')
print('=====>')
print('=====>')
aaa
print('=====>')
print('=====>')
print('=====>')

#2 异常分类
#分类一：针对语法上的异常，应该在程序执行前就解决掉
try:
    print('asdfasdf'
except Exception: #except后有一个万能的异常Exception
    pass
执行结果：还会报错

#分类二：逻辑异常，try...except
# xxx #NameError

# int('xxxxxxx') #valueError

# l=[]
# l[111111] #IndexError

# d={}
# d['a'] #keyError

# 1/0 #ZeroDivisionError:

# import os
# os.xxxxxxxxxxx #AttributeError:

#处理异常的方式：try 。。。except

try:
    print('===>1')
    print('===>2')
    l=[]
    l[123] #IndexError
    print('===>3')
except AttributeError as x:
    print(x)      #打印异常的值
except IndexError as y:
    print(y)
# except Exception as z:   #万能异常
#     print('Ex',z)
else:
    print('被检测的代码块没有发生异常时执行else的代码')
print('====>4')

#finally
try:
    print('===>1')
    print('===>2')
    cursor= connect(数据)
    cursor.excute(sql)
    cursor.excute(sql)
    cursor.excute(sql)
    cursor.excute(sql)
    print('===>3')  #因为上面有异常，此代码不会执行


except Exception:   #万能异常
    print('异常发生时执行的代码')
    # cursor.close()
finally: #不管程序是否出错，都会执行finally的代码
    cursor.close()    #回收资源

#自定义异常
class MySQL_CONN_ERROR(BaseException): #定义一个类，继承父类BaseException
    def __init__(self,value):
        self.value=value

    def __str__(self): #定义报异常的格式
        return '出错啦老铁：%s' %self.value

if 2 > 1:
    # raise TypeError('类型错误')  #自己手动触发一个异常
      raise MySQL_CONN_ERROR('数据库连接错误')

#断言
res=[]
assert len(res) > 0  #如果assert上面的代码有问题，就会报异常，用于调试用
res[0]
res[1]

二、基于udp协议的套接字通讯

前提储备
a,b = (1,2)
print(a)
print(b)
执行结果：
1
2

image.png

#服务端.py
from socket import *

server=socket(AF_INET,SOCK_DGRAM) #SOCK_DGRAM表示udp协议
server.bind(('127.0.0.1',8080))

# server.listen(5) #udp没有连接，不用监听
# server.accept() #udp没有

# while True: #udp没有连接，更不可能有连接循环了
    # server.accept() #udp没有

while True: #通信循环
    msg,client_addr=server.recvfrom(1024) #server.recvfrom(1024)这个函数的返回值是一个元组，里面是(收到消息的内容，发送端的ip和端口)
    print(msg)
    server.sendto(msg.upper(),client_addr)

#客户端.py
from socket import *

client=socket(AF_INET,SOCK_DGRAM)
# client.connect(('127.0.0.1',8080)) #udp没有连接

while True:
    msg=input('>>: ').strip()
    client.sendto(msg.encode('utf-8'),('127.0.0.1',8080))

    msg,server_addr=client.recvfrom(1024)
    print(msg)
总结：udp协议没有连接，发消息时都要指定对方的ip和端口，并且允许发空消息

三、udp协议不会粘包

#服务端.py
from socket import *

server=socket(AF_INET,SOCK_DGRAM) #udp协议叫数据报协议，tcp协议叫流式协议
server.bind(('127.0.0.1',8080))

msg1,client_addr=server.recvfrom(5)  #当收的限制小于5时，windows客户端会报错，linux会少数据
msg2,client_addr=server.recvfrom(1024)
msg3,client_addr=server.recvfrom(1024)

print(msg1)
print(msg2)
print(msg3)

#客户端.py
from socket import *

client=socket(AF_INET,SOCK_DGRAM) #数据报协议

client.sendto('hello'.encode('utf-8'),('127.0.0.1',8080))  #一个sendto就是一个独立的包，服务端也要有一个对应的recvfrom去接收这个包
client.sendto('world'.encode('utf-8'),('127.0.0.1',8080))
client.sendto('egon'.encode('utf-8'),('127.0.0.1',8080))

Tcp和udp的区别
Tcp协议是可靠的协议。因为发送完消息后对方会发送一个确认包，收到确认包后才会从内存中删除这个消息。如果对方没有发确认包，过几秒钟后会重发这个消息，而udp协议，发完就不管了。不管对方收没收到消息都会从内存中删除这个信息，所以udp协议会丢包，但udp协议传输的快

image.png

Udp协议有效传输的数据是512个字节，所以DNS查询的根域只有13台服务器，因为DNS的查询功能是基于udp协议，查询时要包含着13台根域的详细信息，多一台就会超过512个字节

四、进程理论

操作系统的作用提供接口，管理进程

image.png

并发的本质就是保存状态加切换
必备的理论基础：
一 操作系统的作用：
1：隐藏丑陋复杂的硬件接口，提供良好的抽象接口
2：管理、调度进程，并且将多个进程对硬件的竞争变得有序
二 多道技术：
1.产生背景：针对单核，实现并发
ps：
现在的主机一般是多核，那么每个核都会利用多道技术
有4个cpu，运行于cpu1的某个程序遇到io阻塞，会等到io结束再重新调度到这个程序，但会被调度到4个cpu中的任意一个，具体由操作系统调度算法决定。
2.空间上的复用：如内存中同时有多道程序
3.时间上的复用：复用一个cpu的时间片
强调：遇到io切，占用cpu时间过长也切，核心在于切之前将进程的状态保存下来，这样才能保证下次切换回来时，能基于上次切走的位置继续运行
进程：正在进行的一个过程或者说一个任务。而负责执行任务则是cpu。
程序仅仅只是一堆代码而已，而进程指的是程序的运行过程。
进程的创建
一个进程在运行过程中开启了子进程（如nginx开启多进程，os.fork,subprocess.Popen等）

image.png
要提升效率，要让进程尽可能的处于就绪态，减少阻塞态
并发与并行

无论是并行还是并发，在用户看来都是'同时'运行的，不管是进程还是线程，都只是一个任务而已，真是干活的是cpu，cpu来做这些任务，而一个cpu同一时刻只能执行一个任务
一 并发：是伪并行，即看起来是同时运行。单个cpu+多道技术就可以实现并发，（并行也属于并发）
二 并行：同时运行，只有具备多个cpu才能实现并行
单核下，可以利用多道技术，多个核，每个核也都可以利用多道技术（多道技术是针对单核而言的）
有四个核，六个任务，这样同一时间有四个任务被执行，假设分别被分配给了cpu1，cpu2，cpu3，cpu4，一旦任务1遇到I/O就被迫中断执行，此时任务5就拿到cpu1的时间片去执行，这就是单核下的多道技术，而一旦任务1的I/O结束了，操作系统会重新调用它(需知进程的调度、分配给哪个cpu运行，由操作系统说了算)，可能被分配给四个cpu中的任意一个去执行

五、开启子进程的两种方式

##开启子进程的方式一：
from multiprocessing import Process
import time,random

def piao(name):
    print('%s is piaoing' %name)
    time.sleep(random.randint(1,3))
    print('%s is over' %name)

if __name__ == '__main__':   #windows下开子进程一定要导入 '__main__'，记住就可以，只的是此文件做为模块被导入时不会执行下面的代码，只有当做文件运行时才会执行下面的代码
    # p=Process(target=piao,kwargs={'name':'alex'})  #kwargs是以字典的方式给函数传参数
    p=Process(target=piao,args=('alex',))  #args是以位置参数的方式传参数，调用Process类产生一个创建子进程的对象p
    p.start() #仅仅只是向操作系统发送了一个创建子进程p的信号
    print('主')
执行结果：
主
alex is piaoing
alex is over
父进程是给子进程收尸的

#开启子进程的方式二：
from multiprocessing import Process
import time,random

class MyProcess(Process): #定义一个类继承Process类
    def __init__(self,name):
        super(MyProcess,self).__init__()   #继承父类的init方法
        self.name=name

    def run(self):
        print('%s is piaoing' %self.name)
        time.sleep(random.randint(1,3))   #随机的睡[1,3)
        print('%s is over' % self.name)

if __name__ == '__main__':
    p=MyProcess('P1')
    p.start() #仅仅只是向操作系统发送了一个创建子进程p的信号，就会触发对象下的方法run
    print('主')

六、多进程实现并发的套接字通讯

思路：要实现并发，就要创建多个子进程去执行任务
#服务器.py
from socket import *
from multiprocessing import Process

def talk(conn,addr):
    while True:
        try:
            data=conn.recv(1024)
            if not data:break
            conn.send(data.upper())
        except Exception:   #监控异常，如果客户端主动断开会报异常，为了避免报异常需要进行异常处理
            break
    conn.close()

def server():
    s=socket()
    s.bind(('127.0.0.1',8080))
    s.listen(5) #指的是客户端一下子来了好多个，一次只能处理5个连接
    while True:
        conn,addr=s.accept()   #s.accept()函数的返回值是一个元组，（套接字链接,(客户端的ip,端口)）
        print('客户端 %s:%s' %(addr[0],addr[1]))
        p=Process(target=talk,args=(conn,addr))    #创建一个子进程，去干收消息和发消息的活
        p.start()
    s.close()

if __name__ == '__main__':
    server()

# 客户端.py
from socket import *

c = socket()   #默认是tcp协议，不用传参也可以
c.connect(('127.0.0.1', 8080))

while True:
    msg=input('>>: ').strip()
    if not msg:continue
    c.send(msg.encode('utf-8'))
    data=c.recv(1024)
    print(data.decode('utf-8'))

c.close()
这么写有一个问题。就是如果客户端有10000个，就要开一万个子进程去执行任务，这样机器早都崩掉了

七、进程对象的其他方法

#Join方法
from multiprocessing import Process
import time,random

def piao(name):
    print('%s is piaoing' %name)
    time.sleep(random.randint(1,3))
    print('%s is over' % name)


if __name__ == '__main__':
    p1=Process(target=piao,args=('alex1',))
    p1.start()
    p1.join()  #指的是父进程等待子进程执行完之后父进程才结束
    print('主') 
执行结果：
alex1 is piaoing
alex1 is over
主

from multiprocessing import Process
import time,random

def piao(name):
    print('%s is piaoing' %name)
    time.sleep(random.randint(1,3))
    print('%s is over' % name)

if __name__ == '__main__':
    p1=Process(target=piao,args=('alex1',))
    p2=Process(target=piao,args=('alex2',))
    p3=Process(target=piao,args=('alex3',))

    #串行执行
    # p1.start()
    # p1.join()
    # p2.start()
    # p2.join()
    # p3.start()
    # p3.join()

    #并发执行
    p1.start()
    p2.start()
    p3.start()
    p3.join()
    p1.join()
    p2.join()

#简单写法，利用for循环
p_l=[p1,p2,p3] #定义一个列表
for p in p_l:
    p.start()
for p in p_l:
    p.join()
print('主')

#证明进程之间的内存空间是隔离的
from multiprocessing import Process
import time,random

n=100
def task():
    global n #声明是全局变量
    n=0  #改全局变量n的值

if __name__ == '__main__':
    p=Process(target=task)  #产生创建子进程的对象
    p.start() #创建子进程，就会运行task函数，改变全局变量n的值为0
    p.join()  #父进程等待子进程运行完毕后才运行父进程的打印操作
    print('父',n)  #打印父进程内存空间全局变量n的值
执行结果为：父 100
说明创建子进程只是更改自己内存空间中n的值，不会改变父进程中内存空间变量n的值，说明子进程和父进程的内存空间是隔离的

#进程的其他方法和属性
#pid
from multiprocessing import Process
import time,random
import os

def task():
    print('%s is running parent[%s]' %(os.getpid(),os.getppid()))

if __name__ == '__main__':
    p=Process(target=task)
    p.start()
    print(p.pid) #p这个进程的id
    print('主',os.getpid()) #查看python文件的id
    print(os.getppid()) #pycharm的进程id
    time.sleep(1000)

#终止子进程及查看子进程是否存活
from multiprocessing import Process
import time,random
import os

def task():
    print('%s is running ' %(os.getpid()))
    time.sleep(10)

if __name__ == '__main__':
    p=Process(target=task,name='xxxxxxxxxxxxx')
    p.start()
    p.terminate()  #停止子进程，通常不会进行此操作
    time.sleep(1)  #不停止一秒的话，还没来得及干死子进程就查看是否存活，会显示True
    print(p.is_alive())  #查看子进程是否存活
    print(p.name)
    print('主')

八、进程池

#程序的执行方式：
#一：串行执行
#二：并行执行

#同步调用：提交一个任务后，在原地等着，等到该任务运行完毕，拿到结果以后，再执行下一行代码
#异步调用：提交一个任务后，不用在原地等着，直接执行下一行代码，结果呢？
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor,Executor
import time,os,random

def task(i):
    print('%s is running %s' %(os.getpid(),i))
    time.sleep(random.randint(1,3))
    return i**2

if __name__ == '__main__':
    print(os.cpu_count())  #查看cpu的核心数，本机有4个
    pool=ProcessPoolExecutor()  #规定进程池的大小，没有参数默认就是cpu的核心数，进程池是规定同一时间最多可以开多少个进程

    objs=[]
    for i in range(10): #0-9
        obj=pool.submit(task,i) #异步的方式提交任务，会执行task函数，但进程池只有4个，同一时间只有4个进程可以执行任务，obj等于task函数的返回值
        objs.append(obj)  #将task函数的返回值append到列表中

        # res=pool.submit(task,i).result() #同步方式提交任务，会拿到一个结果res，拿到结果后才执行下一个任务
        # print(res)
    pool.shutdown(wait=True) #shutdown代表不允许再往进程池里提交任务,wait=True就是join的意思：父进程等待任务都执行完毕
    print('主')
    for obj in objs:
        print(obj.result())  #最后打印任务的结果
进程池用来开子进程去执行任务

九、paramiko模块的使用

#基于用户名密码连接，远程执行命令
import paramiko  #首先要在parcharm中安装此模块，安装方法

image.png

# 创建SSH对象
ssh = paramiko.SSHClient()
# 允许连接不在know_hosts文件中的主机
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# 连接服务器
ssh.connect(hostname='66.112.215.110', port=8799, username='root', password='xx')

# 执行命令

while True:
    cmd=input('>>: ').strip()
    stdin, stdout, stderr = ssh.exec_command(cmd)  #将命令的执行结果保存到标准输出
    # 获取命令结果
    result = stdout.read()  #充标准输出中读命令的执行结果
    print(result.decode('utf-8'))
 ssh.close() # 关闭连接

#基于密钥登录
import paramiko

private_key = paramiko.RSAKey.from_private_key_file(r'C:\\id_rsa')

# 创建SSH对象
ssh = paramiko.SSHClient()
# 允许连接不在know_hosts文件中的主机
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# 连接服务器
ssh.connect(hostname='123.56.157.199', port=22022, username='root', pkey=private_key)

# 执行命令
stdin, stdout, stderr = ssh.exec_command('df')
# 获取命令结果
result = stdout.read()
print(result.decode('utf-8'))
# 关闭连接
ssh.close()

#上传下载
import paramiko

transport = paramiko.Transport(('123.56.157.199', 22022))
transport.connect(username='root', password='xxx')

sftp = paramiko.SFTPClient.from_transport(transport)
# 将location.py 上传至服务器 /tmp/test.py
sftp.put(r'C:\\id_rsa', '/tmp/test.rsa') #必须写绝对路径
# 将remove_path 下载到本地 local_path
# sftp.get('remove_path', 'local_path')

transport.close()