今日內(nèi)容

---

異常處理

什么是異常?

• 程序在運(yùn)行過(guò)程中出現(xiàn)了不可預(yù)知的錯(cuò)誤
• 并且該錯(cuò)誤沒(méi)有對(duì)應(yīng)的處理機(jī)制,那么就會(huì)以異常的形式表現(xiàn)出來(lái)
• 造成的影響就是整個(gè)程序無(wú)法再正常運(yùn)行

異常的結(jié)構(gòu)

• 1.異常的類(lèi)型:NAMEERROR
• 2.異常的信息:name 'fasdsadsa' is not defined
• 3.異常的位置:

# 異常的位置信息與路徑 Traceback (most recent call last): File ".../.../... 異常處理.py", line1, in<module> fasdsadsa

異常的種類(lèi)

分為兩大類(lèi)

• 1.語(yǔ)法錯(cuò)誤
  • 是你程序立刻就能解決的,這種錯(cuò)誤是不能被容忍的
  • 語(yǔ)法上的錯(cuò)誤,發(fā)現(xiàn)之后應(yīng)該立刻解決
• 2.邏輯錯(cuò)誤
  • 這種錯(cuò)是可以被容忍的,因?yàn)橐谎劭床怀鰜?lái)
  • 針對(duì)邏輯上的錯(cuò)誤,可以采用異常處理機(jī)制進(jìn)行捕獲

常見(jiàn)的錯(cuò)誤類(lèi)型

• NAMERROR 名字錯(cuò)誤
• SyntaxError 語(yǔ)法錯(cuò)誤
• KeyError 鍵不存在
• ValueError 值錯(cuò)誤
• IndexError 索引錯(cuò)誤

錯(cuò)誤類(lèi)型案例:

try:namel = [1,2,3]l[111]d = {'name':'jason'}d['password'] except NameError:print('NameError') except IndexError:print('indexerror') except KeyError:print('keyerror') # 以上錯(cuò)誤依次對(duì)應(yīng)

注意:

• 錯(cuò)誤發(fā)生之后,會(huì)立刻停止代碼的運(yùn)行
• 執(zhí)行except語(yǔ)句,比對(duì)錯(cuò)誤類(lèi)型

萬(wàn)能異常BaseException

try:namel = [1,2,3]l[111]d = {'name':'jason'}d['password']Exception except BaseException: # 萬(wàn)能異常 所有的異常類(lèi)型都被捕獲print('老子天下無(wú)敵')

try>>>else>>>finally

try:namel = [1,2,3]l[111]d = {'name':'jason'}d['password']# except Exception: # 萬(wàn)能異常 所有的異常類(lèi)型都被捕獲print('老子天下無(wú)敵') else:print('被檢測(cè)的代碼沒(méi)有任何的異常發(fā)生 才會(huì)走else') finally:print('無(wú)論被檢測(cè)的代碼有沒(méi)有異常發(fā)生 都會(huì)在代碼運(yùn)行完畢之后執(zhí)行我')

主動(dòng)拋異常以及assert的應(yīng)用

if 'egon' == 'DSB':pass else:raise TypeError('盡說(shuō)大實(shí)話') # 關(guān)鍵字raise就是主動(dòng)拋出異常 ? l = [1,2,3] assert len(l) < 0 # 斷言 預(yù)言 # 猜某個(gè)數(shù)據(jù)的狀態(tài) 猜對(duì)了 不影響代碼執(zhí)行 正常走 # 猜錯(cuò)了 直接報(bào)錯(cuò)

如何避免

• 異常處理
  • 再你認(rèn)為可能會(huì)出現(xiàn)bug的代碼塊上方try一下:注意try內(nèi)部的代碼塊越少越好

• 語(yǔ)法如下:

try:可能出錯(cuò)的代碼 except 出錯(cuò)的類(lèi)型 as e: # 將報(bào)錯(cuò)信息賦值給變量e出錯(cuò)之后的處理機(jī)制

自定義異常

class MyError(BaseException):def __init__(self,msg):super().__init__()self.msg=msgdef __str__(self):return '<dfsdf%ssdfsdaf>' %self.msg raise MyError('我自己定義的異常') # 主動(dòng)拋出異常其實(shí)就是將異常類(lèi)的對(duì)象打印出來(lái),會(huì)走_(dá)_str__方法

---

UDP通信

定義:數(shù)據(jù)報(bào)協(xié)議(自帶報(bào)頭)

總結(jié):

• 1.udp協(xié)議客戶端允許發(fā)空
• 2.udp協(xié)議不會(huì)粘包
• 3.udp協(xié)議服務(wù)端不存在的情況下,客戶端照樣不會(huì)報(bào)錯(cuò)
• 4.udp協(xié)議支持并發(fā)

并發(fā):看起來(lái)像同時(shí)運(yùn)行的

并行:真正意義上的同時(shí)運(yùn)行

服務(wù)端

import socket ? server = socket.socket(type=socket.SOCK_DGRAM) # UDP協(xié)議 server.bind(('127.0.0.1',8080)) # UDP不需要設(shè)置半連接池,它也沒(méi)有半連接池的概念 ? # 因?yàn)闆](méi)有雙向通道,不需要accept,直接就是通信循環(huán) while True:data, addr = server.recvfrom(1024)print('數(shù)據(jù):',data) # 客戶端發(fā)來(lái)的消息print('地址',addr) # 客戶端發(fā)來(lái)的地址server.sendto(data.upper(),addr)

客戶端

import socket ? client = socket.socket(type=socket.SOCK_DGRAM) # 不需要建立連接,直接進(jìn)入通信循環(huán) server_address = ('1227.0.0.1',8080) while True:client.sendto(b'hello',server_address)data, addr = client.recvfrom(1024)print('服務(wù)端發(fā)來(lái)的數(shù)據(jù):',data)print('服務(wù)端發(fā)來(lái)的地址:',addr)

---

關(guān)于socketserver模塊的應(yīng)用

PS:可以讓TCP協(xié)議也支持并發(fā)

TCP協(xié)議服務(wù)端

import socketserver ? class MyServer(socketserver.BaseRequestHandler):def handle(self):while True:data = self.request.recv(1024)print(self.client_address) # 客戶端地址print(data.dncode('utf-8'))self.request.send(data.upper())if __name__ == '__main__':# 創(chuàng)建一個(gè)基于TCP的對(duì)象,只要有客戶端連接,會(huì)自動(dòng)交給自定義類(lèi)中的handle方法去處理socketserver.ThreadingTCPServer(('127.0.0.1',8080),MyServer) # 創(chuàng)建一個(gè)基于TCP的對(duì)象# 啟動(dòng)該服務(wù)對(duì)象server.serve_forever()

TCP協(xié)議客戶端

import socket ? client = socket.socket() client.connect(('127.0.0.1'),8080) ? while True:client.send(b'hello')data = client.recv(1024)print(data.decode('utf-8'))

UDP協(xié)議服務(wù)端

import socketserver ? class MyServer(socketserver.BaseRequestHandler):def handle(self):while True:data,sock = self.requestprint(self.client_address) # 客戶端地址print(data.decode('utf-8'))sock.sendto(data.upper(),self.client_address)if __name__ == '__main__':# 創(chuàng)建一個(gè)基于TCP的對(duì)象,只要有客戶端連接,會(huì)自動(dòng)交給自定義類(lèi)中的handle方法去處理sever = socketserver.ThreadingUDPServer(('127.0.0.1',8080),MyServer) # 創(chuàng)建一個(gè)基于UDP的對(duì)象# 啟動(dòng)該服務(wù)對(duì)象server.serve_forever()

UDP協(xié)議客戶端

import socket import time ? client = socket.socket(type=socket.SOCK_DGRAM) server_address(('127.0.0.1'),8080) ? while True:client.sendto(b'hello',server_address)data,addr = client.recvfrom(1024)print(data.decode('utf-8'),addr)time.sleep(1)

---

用UDP實(shí)現(xiàn)簡(jiǎn)易版本的QQ

服務(wù)端

import socket ? server = socket.socket(type=socket.SOCK_DGRAM) server.bind(('127.0.0.1',8080)) ? while True:data, addr = server.recvfrom(1024)print(data.decode('utf-8'))msg = input('>>>:')server.sendto(msg.encode('utf-8'),addr)

客戶端1

import socket ? client = socket.socket(type=socket.SOCK_DGRAM) server_address = ('127.0.0.1',8080) ? while True:msg = input('>>>:')msg = '來(lái)自客戶端1的消息:%s'%msgclient.sendto(msg.encode('utf-8'),server_address)data, server_addr = client.recvfrom(1024)print(data.decode('utf-8'))

客戶端2

import socket ? client = socket.socket(type=socket.SOCK_DGRAM) server_address = ('127.0.0.1',8080) ? while True:msg = input('>>>:')msg = '來(lái)自客戶端2的消息:%s'%msgclient.sendto(msg.encode('utf-8'),server_address)data, server_addr = client.recvfrom(1024)print(data.decode('utf-8'))

客戶端3

import socket ? client = socket.socket(type=socket.SOCK_DGRAM) server_address = ('127.0.0.1',8080) ? while True:msg = input('>>>:')msg = '來(lái)自客戶端3的消息:%s'%msgclient.sendto(msg.encode('utf-8'),server_address)data, server_addr = client.recvfrom(1024)print(data.decode('utf-8'))

?

---

上傳大型資源類(lèi)文件

服務(wù)端

import socket import os import json import struct ? server = socket.socket() server.bind(('127.0.0.1',8080)) server.listen(5) ? while True:conn,addr = server.accept()while True:try:header_dic = conn.recv(4)# 解析字典報(bào)頭header_len = struct.unpack('i',header_dic)[0]# 再接收字典數(shù)據(jù)header_bytes = conn.recv(header_len)real_dic = json.loads(header_bytes.decode('utf-8'))# 獲取數(shù)據(jù)長(zhǎng)度total_size = real_dic.get('file_size')# 循環(huán)接收并寫(xiě)入文件recv_size = 0with open(real_dic.get('file_name'),'wb') as f:while recv_size < total_size:data = conn.recv(1024)f.write(data)recv_size += len(data)print('上傳成功')except ConnectionResetError as e:print(e)breakconn.close()

客戶端

import socket import json import os import struct ? client = socket.socket() client.connect(('127.0.0.1',8080)) ? while True:# 獲取電影列表 循環(huán)展示MOVIE_DIR = r'絕對(duì)路徑'movie_list = os.listdir(MOVIE_DIR)# print(movie_list)for i,movie in enumerate(movie_list,1):print(i,movie)# 用戶選擇choice = input('please choice movie to upload>>>:')# 判斷是否是數(shù)字if choice.isdigit():# 將字符串?dāng)?shù)字轉(zhuǎn)為intchoice = int(choice) - 1# 判斷用戶選擇在不在列表范圍內(nèi)if choice in range(0,len(movie_list)):# 獲取到用戶想上傳的文件路徑path = movie_list[choice]# 拼接文件的絕對(duì)路徑file_path = os.path.join(MOVIE_DIR,path)# 獲取文件大小file_size = os.path.getsize(file_path)# 定義一個(gè)字典res_d = {'file_name':'性感荷官在線發(fā)牌.mp4','file_size':file_size,'msg':'注意身體,多喝營(yíng)養(yǎng)快線'}# 序列化字典json_d = json.dumps(res_d)json_bytes = json_d.encode('utf-8') ?# 1.先制作字典格式的報(bào)頭header = struct.pack('i',len(json_bytes))# 2.發(fā)送字典的報(bào)頭 client.send(header)# 3.再發(fā)字典 client.send(json_bytes)# 4.再發(fā)文件數(shù)據(jù)(打開(kāi)文件循環(huán)發(fā)送)with open(file_path,'rb') as f:for line in f:client.send(line)else:print('not in range')else:print('must be a number')

轉(zhuǎn)載于:https://www.cnblogs.com/zhukaijian/p/11323492.html

總結(jié)

以上是生活随笔為你收集整理的python day29的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。