今日內容

---

異常處理

什么是異常?

• 程序在運行過程中出現(xiàn)了不可預知的錯誤
• 并且該錯誤沒有對應的處理機制,那么就會以異常的形式表現(xiàn)出來
• 造成的影響就是整個程序無法再正常運行

異常的結構

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

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

異常的種類

分為兩大類

• 1.語法錯誤
  • 是你程序立刻就能解決的,這種錯誤是不能被容忍的
  • 語法上的錯誤,發(fā)現(xiàn)之后應該立刻解決
• 2.邏輯錯誤
  • 這種錯是可以被容忍的,因為一眼看不出來
  • 針對邏輯上的錯誤,可以采用異常處理機制進行捕獲

常見的錯誤類型

• NAMERROR 名字錯誤
• SyntaxError 語法錯誤
• KeyError 鍵不存在
• ValueError 值錯誤
• IndexError 索引錯誤

錯誤類型案例:

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

注意:

• 錯誤發(fā)生之后,會立刻停止代碼的運行
• 執(zhí)行except語句,比對錯誤類型

萬能異常BaseException

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

try>>>else>>>finally

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

主動拋異常以及assert的應用

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

如何避免

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

• 語法如下:

try:可能出錯的代碼 except 出錯的類型 as e: # 將報錯信息賦值給變量e出錯之后的處理機制

自定義異常

class MyError(BaseException):def __init__(self,msg):super().__init__()self.msg=msgdef __str__(self):return '<dfsdf%ssdfsdaf>' %self.msg raise MyError('我自己定義的異常') # 主動拋出異常其實就是將異常類的對象打印出來,會走__str__方法

---

UDP通信

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

總結:

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

并發(fā):看起來像同時運行的

并行:真正意義上的同時運行

服務端

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

客戶端

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

---

關于socketserver模塊的應用

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

TCP協(xié)議服務端

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)建一個基于TCP的對象,只要有客戶端連接,會自動交給自定義類中的handle方法去處理socketserver.ThreadingTCPServer(('127.0.0.1',8080),MyServer) # 創(chuàng)建一個基于TCP的對象# 啟動該服務對象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é)議服務端

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)建一個基于TCP的對象,只要有客戶端連接,會自動交給自定義類中的handle方法去處理sever = socketserver.ThreadingUDPServer(('127.0.0.1',8080),MyServer) # 創(chuàng)建一個基于UDP的對象# 啟動該服務對象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實現(xiàn)簡易版本的QQ

服務端

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 = '來自客戶端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 = '來自客戶端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 = '來自客戶端3的消息:%s'%msgclient.sendto(msg.encode('utf-8'),server_address)data, server_addr = client.recvfrom(1024)print(data.decode('utf-8'))

?

---

上傳大型資源類文件

服務端

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)# 解析字典報頭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ù)長度total_size = real_dic.get('file_size')# 循環(huán)接收并寫入文件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'絕對路徑'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():# 將字符串數(shù)字轉為intchoice = int(choice) - 1# 判斷用戶選擇在不在列表范圍內if choice in range(0,len(movie_list)):# 獲取到用戶想上傳的文件路徑path = movie_list[choice]# 拼接文件的絕對路徑file_path = os.path.join(MOVIE_DIR,path)# 獲取文件大小file_size = os.path.getsize(file_path)# 定義一個字典res_d = {'file_name':'性感荷官在線發(fā)牌.mp4','file_size':file_size,'msg':'注意身體,多喝營養(yǎng)快線'}# 序列化字典json_d = json.dumps(res_d)json_bytes = json_d.encode('utf-8') ?# 1.先制作字典格式的報頭header = struct.pack('i',len(json_bytes))# 2.發(fā)送字典的報頭 client.send(header)# 3.再發(fā)字典 client.send(json_bytes)# 4.再發(fā)文件數(shù)據(jù)(打開文件循環(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')

轉載于:https://www.cnblogs.com/zhukaijian/p/11323492.html

總結

以上是生活随笔為你收集整理的python day29的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。