Websocket通信實戰項目(圖片互傳應用)+PyQt界面+python異步編程(async) （上）服務器端python實現

Rqtz :?個人主頁
???? 共享IT之美，共創機器未來????
?? Sharing the Beauty of IT and Creating the Future of Machines Together

項目背景

?編輯?專有名詞介紹

服務器GUI展示 ?????

功能(位置見上圖序號)

客戶端GUI展示（h5+css+js），平板，手機

動圖?編輯

視頻圖片互傳-CSDN直播

一鍵自動獲取IP地址

websocket通信實現

按鈕映射到新線程啟動websocket服務器

?python中使用async異步實現全雙工通信，B/S主動發送數據，被動接收數據

?圖片二進制轉換顯示到Qt中label控件(涉及到opencv)

上一張，下一張功能實現

整體代碼結構

最后

項目背景

??????? 由于比賽需要，電腦的window系統無法滿足要求，因此就需要安裝linux系統，采用雙系統安裝。安裝完成之后，發現在手機端與電腦端，電腦端和電腦端進行通信(傳輸圖片)時，沒有window上方便，用傳統的方式傳輸的話，大家可能都傾向于QQ或微信，但是在linux上可能就不是那么方便。因此，基于這個問題，我想要開發一個可以在任何平臺都可以運行的圖片互傳軟件，一開始想借助python在各個操作系統上的通用，使用CS架構將服務器和客戶端都編寫成基于Pyqt的Gui界面，但是他們雖然支持在電腦端的各個操作系統之間運行，但是安卓或IOS平臺就不是很優雅，手機不能使用的話，那就失去了方便性。因此，我將服務器端還保持原有的Pyqt的開發方式，在客戶端采用Websocket的方式，原因是websocket支持javascripts編寫，這樣就可以通過瀏覽器來建立客戶端，而瀏覽器在任何平臺都可以使用，包括安卓和IOS平臺，這樣的BS架構就非常優雅的解決了安卓平臺的限制性。同樣這個項目也是對自己學習的一個檢驗。

經過測試，可以在手機端與電腦端，平板端與電腦端，電腦端與電腦端進行全雙工的實時通信！

客戶端實現請看 Websocket通信實戰項目(js)(圖片互傳應用)（下）客戶端H5+css+js實現-CSDN博客

直接看python異步通信async，點擊目錄

?專有名詞介紹

websocket協議：

????????WebSocket是一種實現在單個TCP連接上進行全雙工通信的網絡傳輸協議。這種協議被設計用于改善客戶端和服務器之間實時通信的效率，允許雙方同時發送和接收信息，而無需像傳統HTTP請求那樣輪詢。

CS架構：

????????CS架構則是由客戶端和服務器端組成的兩層結構，客戶端包含業務邏輯和界面展示，服務器端則負責數據管理。這種架構適用于局域網環境，能夠提供快速響應和強大的事務處理能力。CS軟件通常需要專門安裝和維護客戶端程序，因此安全性較高，個性化能力較強。然而，這也導致升級和維護成本較高，且兼容性受限于特定操作系統。

BS架構

????????BS架構是基于瀏覽器和服務器的體系結構，用戶界面通過Web瀏覽器實現，主要業務邏輯在服務器端處理。這種架構使得軟件能夠在不同平臺上運行，客戶端零維護，但個性化能力較低，響應速度相對較慢。由于不需要專門安裝客戶端程序，只需一個網絡鏈接即可訪問，這極大地方便了用戶。然而，BS架構對網絡穩定性要求較高，對硬件的直接支持較弱。????

????

服務器GUI展示 ?????

功能(位置見上圖序號)

點擊按鈕啟動websocket服務器
一鍵自動識別本機 ip地址
圖片接收并顯示在窗口中
圖片數量兩張及以上時，可使用上一戰，下一張切換圖片
支持滾動條，按鈕放大縮小圖片
保存客戶端發送的圖片，支持自定義保存圖片路徑及名稱
在服務器端主動向客戶端發送選擇的圖片，并顯示圖片路徑
必要信息輸出在窗口中，方便觀察。

客戶端GUI展示（h5+css+js），平板，手機

動圖

視頻圖片互傳-CSDN直播

一鍵自動獲取IP地址

????????所謂的自動獲取ip地址，本質上是在終端中輸入查詢ip地址的命令，windows上使用ipconfig，linux(這里是ubuntu)和mac上使用ifconfig，但是使用python要自動獲取，省去了打開終端輸入命令尋找ip的步驟，就需要使用python的os庫，下面請看代碼

def autoip(self):if os.name == 'nt':print("當前操作系統是Windows")output = os.popen("ipconfig | findstr \"IPv4\"").read()ip = output.split("\n")self.myapp.ip.setText(ip[1].split(": ")[1]) elif os.name == 'posix':print("當前操作系統是Linux")output = os.popen("ifconfig | awk '/inet /{print $2}'").read()ip = output.split("\n")self.myapp.ip.setText(ip[1]) elif os.name == 'darwin':print("當前操作系統是Mac")output = os.popen("ifconfig en0 | awk '/inet /{print $2}'").read()self.myapp.ip.setText(output)

1.判斷是哪種操作系統

??? 通過os.name輸出的字符串來判斷是哪種操作系統：

‘nt’ --> Windows系統
‘posix’ --> Linux系統
‘darwin’ --> Mac系統

2.使用os.popen函數獲取命令輸出

windows系統

output = os.popen("ipconfig | findstr \"IPv4\"").read()

解釋：

ipconfig：windows查詢ip地址的命令

“I” :將命令通過管道傳入 findstr命令(windows特有命令)

findstr \"IPv4\"" ：查詢命令輸出中含有IPV4的那一行，注意\"IPv4\"有雙引號

read() : 獲取輸出

?

Linux系統

output = os.popen("ifconfig | awk '/inet /{print $2}'").read()

解釋：

fconfig：linux查詢ip地址的命令

“I” :將命令通過管道傳入 awk命令(linux特有命令)

awk '/inet /{print $2}' 查詢命令輸出中含有IPV4的那一行的第二段字符串

read() : 獲取輸出

打印之后有兩個ip，一個是本地，一個是WLAN，

使用split 函數

ip = output.split("\n")
self.myapp.ip.setText(ip[1])

就可以將ip地址設置到qt的linedit控件中

?

樣例：

?

Mac系統

?output = os.popen("ifconfig en0 | awk '/inet /{print $2}'").read()

查詢指定裝置en0，其他和上述一樣

websocket通信實現

按鈕映射到新線程啟動websocket服務器

1.將按鈕通過信號和曹連接到啟動新線程函數中

#初始化信號和槽
self.myapp.start.clicked.connect(self.newprocess)

newprocess為啟動新的線程的函數

2.啟動子線程函數newprocess實現

#啟用子線程def newprocess(self):if self.myapp.port.text() == "" or self.myapp.ip.text() == "":self.myapp.picdata.append("【"+str(time.time())+"】"+"【錯誤】："+"請輸入端口或者ip地址")else:th = threading.Thread(target=self.connect_server)th.start()self.myapp.start.setDisabled(True)

解釋：

（1）? if self.myapp.port.text() == "" or self.myapp.ip.text() == "":

判斷端口輸入框和ip地址輸入框是否為空，為空則發出警告，

（2）?th = threading.Thread(target=self.connect_server)? ??? th.start()

不為空則可以用使用threading函數來創建一個線程啟動websocket服務器。

（3）self.myapp.start.setDisabled(True)

?????? 啟動成功則可以將按鈕設置為不能點擊，防止重復啟動服務器

問題

????????為什么要用一個新的線程呢？因為websocket服務器啟動時，會阻塞當前線程，當前有一個主線程用于GUI界面的交互(鼠標點擊按鈕，拖動頁面等)，如果服務器在主線程啟動，且一直沒有客戶端連接的話，界面就會卡死，所有按鈕都無法點擊，因為主線程阻塞。所以要用一個新的線程啟動服務器。

2.子線程函數connect_server實現，異步，協程

#初始化websocket服務器，異步def connect_server(self):self.emitdata.emit("【提示】："+"服務器監聽中")loop = asyncio.new_event_loop()asyncio.set_event_loop(loop)self.start_server = websockets.serve(self.handler,self.myapp.ip.text(), 8899,max_size=7000000)loop.run_until_complete(self.start_server)loop.run_forever()
#在顯式的stop事件循環后,取消所有任務for task in asyncio.all_tasks(loop):  task.cancel()print(task.cancelled())loop.close()

解釋：

（1）
self.emitdata.emit("【提示】："+"服務器監聽中")
????????由于connect_server是一個子線程，子線程中無法直接訪問主線程，emitdata是自定義的信號,通過在合適的位置發射信號,再連接到特定的函數中.

????????但是我的ui元素是定義在類的self屬性中的,子線程可以直接通過self直接訪問它,但是經過測試發現我們在子線程中直接向QTextEdit中增添數據時,會報錯QObject::connect: Cannot queue arguments of type 'QTextCursor',因此最好還是使用信號和槽的方式來進行主子線程通信.

（2）
 loop = asyncio.new_event_loop()  asyncio.set_event_loop(loop)
使用python中asyncio庫新建一個event_loop事件循環，并且將新創建的事件循環設置為當前線程的事件循環，事件循環是處理異步操作的核心組件

（3）
 self.start_server = websockets.serve(self.handler,self.myapp.ip.text(), 8899,max_size=7000000)
為什么是異步的呢，因為websocket服務器的回調函數self.handler必須是一個異步函數，

websockets.serve函數的參數

第一個參數：服務器連接成功后調用的函數，必須是異步的
第二個參數：ip地址
第三個參數：端口號
第四個參數：傳輸的最大字節數
還有其他可選參數，這里只用了四個

（4）
 loop.run_until_complete(self.start_server)  loop.run_forever()
運行傳入的協程self.start_server，并讓事件循環一直運行下去，self.start_server是一個協程

（5）bug
#在顯式的stop事件循環后,取消所有任務for task in asyncio.all_tasks(loop):  task.cancel()print(task.cancelled())loop.close()
由于loop.run_forever()會讓事件循環一直運行下去,期間會阻塞線程,直到顯式的使用stop()方法,這個stop方法的調用是在異步發送數據函數中通過捕捉closeflag標志位來實現的.然后在但前的事件循環中取消所有任務,但是發現有個報錯,我一直都沒有解決,下文也有提到 ???????

??值的是task4被取消但是仍在掛起狀態,這個task4是一個WebSocketServerProtocol.handler().

?python中使用async異步實現全雙工通信，B/S主動發送數據，被動接收數據

websocket回調函數

 #websocket處理函數async def handler(self,websocket,path):#創建兩個task，分別為發送和接收sendtask = asyncio.get_event_loop().create_task(self.send(websocket))receivetask = asyncio.get_event_loop().create_task(self.receive(websocket))#異步執行await sendtaskawait receivetask

異步發送數據函數

    #異步發送數據async def send(self,websocket):while True:#點擊發送圖片按鈕后，標志位為真if self.sendflag:#此時的self.curr_bytedata存儲的二進制數據為選擇的圖片await websocket.send(self.curr_bytedata)self.sendflag = Falseself.emitdata.emit("發送成功！")#點擊斷開連接按鈕后if self.closeflag:#關閉websocketawait websocket.close()#顯式的停止事件循環loop =  asyncio.get_event_loop()loop.stop()#跳出循環,終止協程break#掛起1s，切換到其他協程await asyncio.sleep(1)

異步接收數據函數

    async def receive(self,websocket):self.emitdata.emit(f"客戶端連接成功，連接到{websocket.remote_address}")try:async for message in websocket:self.curr_bytedata = message#字節大小print(len(message))self.show_image()except  websockets.ConnectionClosedError:self.emitdata.emit("客戶端意外斷開連接，請客戶端重連")

解釋

（1）websocket回調函數handler

由于websocket服務器的回調函數必須是一個異步函數self.handler,因此該函數必須加上async前綴，才可以將其變成一個協程。當服務器檢查到有客戶端鏈接過來時就會調用這個回調函數handler。
????????當客戶端連接后，self.handler將傳入以下兩個主要參數

websocket：這是一個websockets.WebSocketServerProtocol實例，它代表服務器端與客戶端之間的WebSocket連接。通過這個對象，您可以發送和接收WebSocket幀。
path：這是一個字符串，表示請求的URL路徑。對于WebSocket服務器來說，這個值通常是/，但理論上可以是任何值，取決于如何配置websockets.serve函數。
?
sendtask = asyncio.get_event_loop().create_task(self.send(websocket))
receivetask = asyncio.get_event_loop().create_task(self.receive(websocket))       
?????????創建出兩個task，并且把他們設置到當前的事件循環中。
?
await sendtask      
await receivetask       
????????? await 關鍵字，后面必須跟上一個可等待的對象，例如task，future等，這里面的send和receive就是task對象，使用await關鍵字就可以將控制權交給evet_loop事件循環。
注意async def為前綴的函數是一個異步函數，必須把它放到事件循環中才可以運行，如果像以往那樣子直接調用函數是不會執行的，而是返回一個coroutine對象。
（2）異步發送數據函數與異步接收數據函數

??????? 在網上找到的資料幾乎全部都是在服務器受到客戶端消息時才向客戶端發消息，但是我這個的話，發送圖片的這個操作完全是有用戶自主決定的，即用戶想什么時候發送就什么時候發送，如果只在服務器收到消息才發的話，那也太沒意思了。那即要求發送又要求實時接收，首先循環是必要的，但是通信過程中用戶并不是時時刻刻在發送，也不是時時刻刻在接收，因此大多數時間都是在等待的，因此我們需要使用異步休眠的方式在發送的協程和接收的協程之間不斷的切換，在await等待的過程中做別的事情，以提高程序的效率。
?
  while True:if self.sendflag: ......await asyncio.sleep(1)
?????????首先創建了一個死循環，不斷的判斷用戶有沒有點擊發送按鈕，即sendflag有沒有變為真，判斷結束后，?await asyncio.sleep(1)，異步休眠一秒，這里休眠的作用是可以暫停該協程一秒，來去切換到其他協程，剛剛說了await可以將控制權交給事件循環，事件循環此時就檢查當前還有哪些任務可以執行，發現還有一個receivetask可以執行，因此就利用這一秒鐘的時間切換到這個receivetask協程，這也就是為什么服務器連接成功后會在窗口打印“客戶端連接成功“，因為利用了這一秒鐘執行了receivetask協程中的self.emitdata.emit(f"客戶端連接成功，連接到{websocket.remote_address}")。
 try:async for message in websocket:self.curr_bytedata = messageprint(len(message)) #字節大小self.show_image()except  websockets.ConnectionClosedError:self.emitdata.emit("客戶端意外斷開連接，請客戶端重連")              
?????? 接著進行try，async for message in websocket將會從websocket中檢查有無數據，? 注意：這也是一個異步的對象，也使用的async for，也會將控制權交給事件循環，如果此時客戶端沒有發數據的話，事件循環就會檢查當前還有哪些協程可以執行，于是又切會sendtask協程，其實我認為1s之后還是會切換回去。總的來說，這個for循環是只要有數據發來就執行，沒數據就等待，這個等待可以切換到別的協程中。
?????? 如果此時客戶端發來數據時，會將發來的圖片的二進制數據，賦值給一個變量，在經過self.show_image()處理顯示。下方會有介紹
except 檢查報錯。
如果用戶點擊了發送按鈕，即self.sendflag 為真，
if  self.sendflag:#此時的self.curr_bytedata存儲的二進制數據為選擇的圖片await websocket.send(self.curr_bytedata)self.sendflag = Falseself.emitdata.emit("發送成功！")
??? 我們將當前用戶選擇的圖片的二進制格式的數據發送給客戶端，使用send方法，發送完成后，該task就結束了，一般很短時間內就發送完成，取決于網絡，然后重新將標志位標為假，等待用戶下一次點擊。
????? (3)? 關閉連接和停止事件循環(bug)
 #點擊斷開連接按鈕后if self.closeflag:#關閉websocketawait websocket.close()#顯式的停止事件循環loop =  asyncio.get_event_loop()loop.stop()#跳出循環,終止協程break
????????關閉連接后,關閉websocket連接,此時receivetask中的異步循環由于斷開了連接,該任務終止,sendtask在關閉連接之后break跳出了循環,sendtask也終止,顯式的stop事件循環,最后在conncect_server函數最后有取消掉了沒有關閉的任務,但是顯示取消失敗,并附帶報錯,和上文提到的bug是同一個,也就是self.handler無法取消,task.canceled()返回false,?我也不知道為什么.希望能看出問題的大佬解答!

?圖片二進制轉換顯示到Qt中label控件(涉及到opencv)

?????? show_image()顯示圖像函數實現

 #顯示圖像def show_image(self):。binarydata = np.frombuffer(self.curr_bytedata,np.uint8)self.image = cv2.imdecode(binarydata,cv2.IMREAD_COLOR)value = cv2.cvtColor(self.image,cv2.COLOR_BGR2RGB)height, width, channels = self.image.shapeimages = QImage(value.data, width, height, width * channels, QImage.Format_RGB888)
#顯示圖片 self.myapp.image.setPixmap(QPixmap.fromImage(images).scaled(int(width/self.scale_percent),int(height/self.scale_percent)))

解釋
binarydata = np.frombuffer(self.curr_bytedata,np.uint8)
將二進制數據self.curr_bytedata轉換為NumPy數組，數據類型為np.uint8。
 self.image = cv2.imdecode(binarydata,cv2.IMREAD_COLOR)
將二進制數據解碼為圖self.image，解碼格式為彩色（cv2.IMREAD_COLOR）。
 value = cv2.cvtColor(self.image,cv2.COLOR_BGR2RGB)
將圖像從BGR格式轉換為RGB格式
height, width, channels = self.image.shape
獲取圖像的高度、寬度和通道數
images = QImage(value.data, width, height, width * channels, QImage.Format_RGB888)
轉換成QImage在 ui上顯示
 self.myapp.image.setPixmap(QPixmap.fromImage(images).scaled(int(width/self.scale_percent),int(height/self.scale_percent)))
顯示圖片，在label控件上

上一張，下一張功能實現

排除重復圖片

根據下面代碼，在show_image中添加

# 轉換成QImage在 ui上顯示
#images = QImage(value.data, width, height, width * channels, QImage.Format_RGB888)#中間插入下面的flag = False#將每次顯示的不同的圖像加入imagelist列表中，為按鈕切換上，下張準備for k in range(len(self.imagelist)):if self.curr_bytedata == self.imagelist[k]:flag = Trueif flag == False:self.imagelist.append(self.curr_bytedata)self.number = len(self.imagelist)#圖片為2張及以上時使能上一張下一張按鈕if self.number > 1:self.myapp.up.setDisabled(False)self.myapp.down.setDisabled(False)#中間插入上面的
#顯示圖片 
#self.myapp.image.setPixmap(QPixmap.fromImage(images).scaled(int(width/self.scale_percent),int(height/self.scale_percent)))

解釋
 for k in range(len(self.imagelist)):if self.curr_bytedata == self.imagelist[k]:flag = True
??????? 首先flag初始為假，這個for循環是指在存儲圖片二進制數據的imagelist列表中遍歷當前的圖片歷表中是否有重復的，有的話flag為真。??????????????????????????????????????? ??????????????????
if flag == False:self.imagelist.append(self.curr_bytedata)self.number = len(self.imagelist)
?? 當當前的圖片數據沒有和之前的重復時，就往該列表imagelist中追加新的數據，self.number為這個列表的長度。 ?????????????????????????????????????????????????????????????????????????
if self.number > 1:self.myapp.up.setDisabled(False)self.myapp.down.setDisabled(False)
??? 圖片為2張及以上時使能上一張下一張按鈕

上，下一張按鈕實現

上一張

#上一張def uppic(self):self.number -= 1if self.number  < 1:self.number = len(self.imagelist)self.curr_bytedata = self.imagelist[self.number-1]self.show_image()else:self.curr_bytedata = self.imagelist[self.number-1]self.show_image()

下一張

#下一張def downpic(self):self.number += 1if self.number > len(self.imagelist):self.number = 1self.curr_bytedata = self.imagelist[self.number-1]self.show_image()else:self.curr_bytedata = self.imagelist[self.number-1]self.show_image()