PyQt5+socket+多线程，实现简单聊天室

之前从没想过自己设计一套网络聊天室出来，但网络原理实验的时候老师说:"谁做出来了,就给谁加分",于是我（内卷王）又开始(偷偷地背着全班同学)自己做起了这一个实验项目。实验项目的主要要求简单明了，就是"能互相通信"，"有界面"，说着轻巧，做起来可不简单，花了我三天时间才肝出来。

1.聊天室功能:

2.聊天室的实现原理

3.连接测试

3.1 本地连接测试

3.2 跨网段连接测试

4.聊天室关键代码解释

5.完整代码

6.大坑提醒

7.结束语

1.聊天室功能:

(1) 不同客户端之间能够通过服务端转发实现相互通信

(2) 客户端和服务端均采用多线程处理工作

(3) 具有简单明了，排版整洁的图形化界面

2.聊天室的实现原理

socket库:基于socket库提供的远程通信函数，实现远程通信，基础函数如下:
socket.socket(socket.AF_INET,socket.SOCK_STREAM)
socket.AF_INET # 使用IPv4
socket.SOCK_STREAM # 基于TCP的提供有保障的面向连接的服务
设定完通信协议后服务端使用bind函数绑定本机IP和端口(即套接字)
socket.bind((ip,port)) # ip为服务端ip地址，端口为设定的开放端口
客户端指定好ip和端口后通过connect函数与服务端建立连接:
socket.connet((ip,port))
连接成功后客户端和服务端可以互相通过send和recv函数发送和接受消息。
==============================================================
多线程处理:基于PyQt5自带的QThread类和特有且好用的pyqtSignal
基于PyQt5提供的QThread类为服务端开通多个客户端监听线程，pyqtSignal负责实现多个线程与主线程的异步消息传输。一开始服务端主线程开启一个监听线程监听客户端连接请求并维持会话，一旦有监听线程连接到客户端后开启一个新的监听线程监听下一个可能存在的客户端请求。当监听线程接收到各自的客户端发来的消息之后，使用pyqtSignal向主线程发送信号，主线程接收到信号后向全体客户端广播内容。

3.连接测试

3.1 本地连接测试

为什么?要是本地连接测试都不行，就别指望不同主机之间连接了....，首先分别设定好服务端和客户端输入IP和端口后,服务端先点击建立服务器，然后客户端在点击连接服务器。

连接成功后客户端向服务端发送消息，服务端向客户端回复消息，连接测试成功!!!

3.2 跨网段连接测试

在腾讯云上租了一个服务器，然后将自己的服务端代码放在了上面,使用两台电脑连接外网服务器并相互通信。(服务器IP就不透露了)

不同网段之间的连接测试成功!!!

4.聊天室关键代码解释

服务端实例(主线程)

广播消息函数：当接受到消息的时候触发该函数，通过遍历线程列表依次将消息按顺序发送给每个客户端，如果发送失败就断开与该客户端的连接

def bordCastInfo(self,info):print(len(self.serverDict))for client in self.serverDict:try:if self.serverDict[client].clientsocket != None:print("尝试将信息广播出去")self.serverDict[client].sendToClient(info)     # 将消息传入指定的客户端print("广播成功")except Exception as reason:self.getFlag("@@@".join([client,"disconnect"]))                       # 运行函数,停止某个客户端的监听(相当于关闭)print("服务端",reason)

连接状态signal绑定的槽函数,如果发来的消息是成功的，则开启一个新的监听线程线程，否则设定监听线程运行转态为False,然后子线程自主结束。

def getFlag(self,flag):flag = flag.split("@@@")if flag[1] == "connect":    # 如果传来连接成功,则新开一个线程监听self.buildServer()elif flag[1] == "disconnect": # 如果连接出现问题self.serverDict[flag[0]].runflag = False

监听线程实例(子线程)：

消息持续接受函数:循环尝试接受消息,接收到则调用sendText函数发送消息给主线程(主线程负责在界面上显示消息)，如果失败则发送失败原因给主线程(主线程接受并显示在状态栏上)，发送连接失败文本，发送连接失败状态，然后主动关闭线程

def getMessage(self):while self.runflag:try:data = self.clientsocket.recv(1024).decode('utf-8')   # 接受到字符串并按照utf-8编译self.sendText(data)except Exception as reason:self.sendMessage(str(reason))self.sendText(str(self.addr)+" break connect...")self.sendFlag(1)       # 发送断开连接标志breakself.clientsocket.close()print("线程关闭成功")

消息发送函数:由主线程依次调用不同子线程的发送函数发送。

def sendToClient(self,info):print(self.clientsocket)try:self.clientsocket.send(info.encode("utf-8"))print("广播成功")except Exception as reason:print("广播失败原因",reason)self.sendMessage(self.addr+" break connect...")self.sendFlag(1)

客户端子线程实例:

启动子线程之后自动执行run函数,持续接受消息，作用与服务端监听线程实例的getMessage函数差不多

def run(self):while self.runflag:try:msg = self.serverSocket.recv(1024).decode("utf-8")   # 接受服务端消息self.sendText(msg)except Exception as reason:self.sendMessage(reason)self.sendFlag(1)                      # 发送连接失败标志break

5.完整代码

这个代码主要使用在有图形化界面的系统上,可以即当服务端也可以当客户端，租来的服务器没有图形化界面可用不了...

from PyQt5.QtWidgets import *
from PyQt5.QtCore import *
from PyQt5.QtGui import *
from threading import Thread
import socket
import sys# 页面实例
class MainWin(QMainWindow):def __init__(self,parent = None):super(MainWin,self).__init__(parent)self.pc = None                            # 预设这个变量self.MainSet()self.createWidget()self.componentWidgets()# 设定Main窗口属性的函数def MainSet(self):self.setWindowTitle("SimpleChat!!")#self.setFixedSize(600,400)# 加载控件def createWidget(self):self.centerWidget = QWidget()                    # 中心窗口控件self.mainLayout = QGridLayout()self.rightTopLayout = QGridLayout()self.rightBottomLayout = QVBoxLayout()self.ConfigBox = QGroupBox("Configuartion")self.ContralBox = QGroupBox("Contral Panel")self.chatEdit = QTextEdit()                        # 聊天对话框self.inputLine = QLineEdit()                    # 输入框self.sendBtn = QPushButton("发送")                # 发送按钮self.sendBtn.clicked.connect(self.sendInfo)       # 向服务器发送信息(如果是服务器本身则广播)self.clearBtn = QPushButton("清空")             # 清空按钮self.clearBtn.clicked.connect(lambda:self.inputLine.clear())self.ipEdit = QLineEdit()                        # IP输入栏self.ipEdit.setInputMask('000.000.000.000; ')  # 设定为IP输入格式self.hostIPbtn = QPushButton("获得本机IP")        # 点击直接获得本机IPself.hostIPbtn.clicked.connect(self.getHostIP)self.portEdit = QLineEdit()                      # 端口输入栏self.portEdit.setPlaceholderText("9999")       # 默认端口为9999self.hostEdit = QLineEdit()                     # 主机名输入栏self.hostEdit.setPlaceholderText(socket.gethostname())  # 默认为本地主机self.serverRbtn = QRadioButton("Server")        # 选择为服务器self.serverRbtn.setChecked(True)self.serverRbtn.toggled.connect(self.radiobtnChange)self.clientRbtn = QRadioButton("Client")     # 选择为客户端self.connectBtn = QPushButton("连接服务器")       # 连接服务器按钮self.connectBtn.clicked.connect(self.setClient)self.connectBtn.setEnabled(False)self.buildServerBtn = QPushButton("建立服务器")  # 建立服务器按钮self.buildServerBtn.clicked.connect(self.setServer)self.quitBtn = QPushButton("退出")             # 退出按钮self.quitBtn.clicked.connect(self.quit)# ---------------------------------------------------------self.statusBar = QStatusBar()                  # 状态栏# 组装控件def componentWidgets(self):self.setCentralWidget(self.centerWidget)self.setStatusBar(self.statusBar)# --------------------------------------------------------self.centerWidget.setLayout(self.mainLayout)self.mainLayout.addWidget(self.chatEdit,0,0,6,2)self.mainLayout.addWidget(self.inputLine,6,0,1,2)self.mainLayout.addWidget(self.sendBtn,7,0,1,1)self.mainLayout.addWidget(self.clearBtn,7,1,1,1)self.mainLayout.addWidget(self.ConfigBox,0,2,5,1)self.mainLayout.addWidget(self.ContralBox,5,2,3,1)self.ConfigBox.setLayout(self.rightTopLayout)self.rightTopLayout.addWidget(QLabel("Server IP"),0,0,1,4)self.rightTopLayout.addWidget(self.ipEdit,1,0,1,3)self.rightTopLayout.addWidget(self.hostIPbtn,1,3,1,1)self.rightTopLayout.addWidget(QLabel("Server Port"),2,0,1,1)self.rightTopLayout.addWidget(self.portEdit,2,1,1,3)self.rightTopLayout.addWidget(QLabel("Host Name"),3,0,1,1)self.rightTopLayout.addWidget(self.hostEdit,3,1,1,3)self.rightTopLayout.addWidget(self.serverRbtn,4,0,1,2)self.rightTopLayout.addWidget(self.clientRbtn,4,2,1,2)self.ContralBox.setLayout(self.rightBottomLayout)self.rightBottomLayout.addWidget(self.connectBtn)self.rightBottomLayout.addWidget(self.buildServerBtn)self.rightBottomLayout.addWidget(self.quitBtn)# 静置函数 - 用于写事件函数-------------------------# 单选按钮切换函数def radiobtnChange(self,status):if status:self.connectBtn.setEnabled(False)self.buildServerBtn.setEnabled(True)else:self.connectBtn.setEnabled(True)self.buildServerBtn.setEnabled(False)def getHostIP(self):hostip = socket.gethostbyname_ex(socket.gethostname())self.ipEdit.setText(hostip[-1][-1])# 状态栏情况发送函数def sendInfo(self):if self.pc == None:self.statusBar.showMessage("sned info field case out connected!!")else:info = self.inputLine.text()if info != "":info = self.pc.hostName+":\n"+infoself.pc.btnsend(info)else:self.statusBar.showMessage("input can't be none!")# 设定本主机为服务器def setServer(self):host = self.hostEdit.text()port = self.portEdit.text()ip = self.ipEdit.text()print(ip)if host == "":host = "服务管理员"            # 服务主机if port == "":port = 9999                    # 默认端口if ip == "...":ip = "127.0.0.1"                # 默认IPself.pc = Server(self,ip,host,int(port))# 设定本主机为客户端def setClient(self):host = self.hostEdit.text()port = self.portEdit.text()ip = self.ipEdit.text()if host == "":host = "匿名用户"                # 匿名用户if port == "":port = 9999                    # 默认端口if ip == "...":ip = "127.0.0.1"                # 默认IPself.pc = Client(self,ip,host,int(port))def quit(self):if self.pc != None:self.pc.closeThread()self.close()# 服务端
class Server():def __init__(self,widget,ip,host,port):# 设定本主机的一些基本信息 ---------------------------------------self.widget = widgetself.ip = ip                              # 获得该主机ipself.hostName = host                      # 获得该主机名self.port = port                           # 设定默认端口号(服务器端口号和客户端接入端口号都是这个默认端口)self.serverDict = {}                     # 服务线程字典self.serverID = 0                          # 初始的服务线程idself.buildSocket()# 创建网络连接实例def buildSocket(self):self.socket = socket.socket(socket.AF_INET,socket.SOCK_STREAM)# socket.AF_INET(!!INET - IPv4  INET6 - IPv6)# socket.SOCK_STREAM - 传输控制协议(TCP)self.initialServer()# 初始化服务(仅限服务器)def initialServer(self):# 首先绑定服务端口号self.socket.bind((self.ip,self.port))       # 绑定端口与主机名self.socket.listen(5)                         # 设定最大连接数self.buildServer()                             # 初始化一个服务线程# 创建服务线程def buildServer(self):server = ServerThread(str(self.serverID),self.socket)self.serverDict[str(self.serverID)] = serverself.serverID+=1server._flag.connect(self.getFlag)server._signal.connect(self.getMessage)server._text.connect(self.getText)server.start()# 广播所有消息def bordCastInfo(self,info):print(len(self.serverDict))for client in self.serverDict:try:if self.serverDict[client].clientsocket != None:print("尝试将信息广播出去")self.serverDict[client].sendToClient(info)     # 将消息传入指定的客户端print("广播成功")except Exception as reason:self.getFlag("@@@".join([client,"disconnect"]))                       # 运行函数,停止某个客户端的监听(相当于关闭)print("服务端",reason)def btnsend(self,text):self.widget.chatEdit.append(text)self.bordCastInfo(text)# 关闭所有的服务def closeThread(self):for server in self.serverDict:self.serverDict[server].runflag = Falsedef getFlag(self,flag):flag = flag.split("@@@")if flag[1] == "connect":  # 如果传来连接成功,则新开一个线程监听self.buildServer()elif flag[1] == "disconnect": # 如果连接出现问题self.serverDict[flag[0]].runflag = Falsedef getMessage(self,signal):signal = signal.split("@@@")self.widget.statusBar.showMessage("serverID "+signal[0]+" status:"+signal[1])def getText(self,text):self.widget.chatEdit.append(text)self.bordCastInfo(text)                    # 广播出去# 客户端
class Client(): # 主机默认为本地主机,def __init__(self,widget,ip,hostName,port):self.widget = widgetself.ip = ipself.hostName = hostName                  # 获得该主机名self.port = port                           # 设定默认端口号(服务器端口号和客户端接入端口号都是这个默认端口)self.buildSocket()def buildSocket(self):self.socket = socket.socket(socket.AF_INET,socket.SOCK_STREAM)self.buildClient()def buildClient(self):self.client = ClientThread(self.socket)       # 获取连接self.client._flag.connect(self.getFlag)self.client._signal.connect(self.getMessage)self.client._text.connect(self.getText)if self.client.connectServer(self.ip,self.port):self.client.start()def sendToServer(self,text):                 # 向服务器发送消息try:self.socket.send(text.encode('utf-8'))except Exception as reason:self.getMessage(reason)self.getFlag("disconnect")                # 发送连接失败标志def btnsend(self,text):self.sendToServer(text)def closeThread(self):self.runflag = Falsedef getFlag(self,flag):if flag == "connect":self.widget.statusBar.showMessage("connect success!!")elif flag == "disconnect":self.client.runflag = Falsedef getMessage(self,signal):self.widget.statusBar.showMessage(signal)def getText(self,text):self.widget.chatEdit.append(text)# 监听连接线程,负责构成会话(服务端线程)
class ServerThread(QThread):_signal = pyqtSignal(str)                              # 设定信号,主要向主线程发送信号_text = pyqtSignal(str)                               # 设定信号,向主线程发送接收到的信息_flag = pyqtSignal(str)                             # 设定信号,向主线程发送连接状态标志def __init__(self,serverID,serverSocket):super(ServerThread, self).__init__()self.serverID = serverID           # 获得主机实例self.serverSocket = serverSocketself.clientsocket = Noneself.addr = Noneself.runflag = Trueself.connectList = ["connect","disconnect"]     # 连接成功与连接失败#自动进行该函数def run(self):self.sendMessage("Waiting for customer......")self.clientsocket,self.addr = self.serverSocket.accept()  # 收到客户端的连接后返回 连接控件,地址(持续监听,直到接收到执行下一个操作)print(self.clientsocket)self.sendText("Customer IP: %s" % str(self.addr)+" is linking!")self.sendFlag(0)   # 发送连接成功标志self.getMessage()# 持续接受消息def getMessage(self):while self.runflag:try:data = self.clientsocket.recv(1024).decode('utf-8')   # 接受到字符串并按照utf-8编译self.sendText(data)except Exception as reason:self.sendMessage(str(reason))self.sendText(str(self.addr)+" break connect...")self.sendFlag(1)       # 发送断开连接标志breakself.clientsocket.close()print("线程关闭成功")def sendToClient(self,info):print(self.clientsocket)try:self.clientsocket.send(info.encode("utf-8"))print("广播成功")except Exception as reason:print("广播失败原因",reason)self.sendMessage(self.addr+" break connect...")self.sendFlag(1)# 发送状态信号函数def sendMessage(self,message):self._signal.emit("@@@".join([self.serverID,message]))# 发送接收到的消息信号def sendText(self,text):self._text.emit(text)# 发送连接状态标志def sendFlag(self,flagIndex):self._flag.emit("@@@".join([self.serverID,self.connectList[flagIndex]]))class ClientThread(QThread):_signal = pyqtSignal(str)_text = pyqtSignal(str)_flag = pyqtSignal(str)def __init__(self,serverSocket):super(ClientThread,self).__init__()self.serverSocket = serverSocketself.runflag = Trueself.connectList = ["connect","disconnect"]       # 连接成功与连接失败def connectServer(self,ip,port):try:self.serverSocket.connect((ip,port))self.sendFlag(0)                         # 发送连接成功标志return Trueexcept Exception as reason:self.sendMessage(reason)self.sendFlag(1)                            # 发送链接失败标志return reasondef run(self):while self.runflag:try:msg = self.serverSocket.recv(1024).decode("utf-8")   # 接受服务端消息self.sendText(msg)except Exception as reason:self.sendMessage(reason)self.sendFlag(1)                      # 发送连接失败标志breakdef sendMessage(self,message):self._signal.emit(str(message))def sendText(self,text):self._text.emit(str(text))def sendFlag(self,flagIndex):self._flag.emit(str(self.connectList[flagIndex]))if __name__ == "__main__":app = QApplication(sys.argv)win = MainWin()win.show()sys.exit(app.exec_())

服务器服务端代码,这个主要是我用在腾讯云服务器上的代码，因为服务器Ubuntu没有界面,就省去了界面制作。!!在ubuntu上运行的python如果写了中文得表明为gbk编码,否则会报错。其次这里主要使用的是PySide2(一个和PyQt5基本一样的模块),因为PyQt5在ubuntu上安装半天都失败。

# -*- coding:gbk -*-
#!/usr/bin/python3
from PySide2.QtWidgets import QApplication
from PySide2.QtCore import QThread,Signal
import socket
import sysclass Server():def __init__(self,ip,host,port):self.ip = ipself.hostName = hostself.port = portself.serverDict = {}self.serverID = 0self.buildSocket()def buildSocket(self):self.socket = socket.socket(socket.AF_INET,socket.SOCK_STREAM)self.initialServer()def initialServer(self):self.socket.bind((self.ip,self.port))self.socket.listen(5)print("start create serverThread")self.buildServer()def buildServer(self):server = ServerThread(str(self.serverID),self.socket)print("Thread server Created!")self.serverDict[str(self.serverID)] = serverself.serverID+=1server._flag.connect(self.getFlag)server._signal.connect(self.getMessage)server._text.connect(self.getText)server.start()print("Server create successfully!")def bordCastInfo(self,info):for client in self.serverDict:try:if self.serverDict[client].clientsocket != None:print("try boardcasting")self.serverDict[client].sendToClient(info)print("boardcast success")except Exception as reason:self.getFlag("@@@".join([client,"disconnect"]))print("Server:",reason)def btnsend(self,text):self.widget.chatEdit.append(text)self.bordCastInfo(text)def closeThread(self):for server in self.serverDict:self.serverDict[server].runflag = Falsedef getFlag(self,flag):flag = flag.split("@@@")if flag[1] == "connect":self.buildServer()elif flag[1] == "disconnect":self.serverDict[flag[0]].runflag = Falsedef getMessage(self,signal):signal = signal.split("@@@")print("serverID "+signal[0]+" status:"+signal[1])def getText(self,text):print(text)self.bordCastInfo(text)class ServerThread(QThread):_signal = Signal(str)_text = Signal(str)_flag = Signal(str)def __init__(self,serverID,serverSocket):super(ServerThread, self).__init__()self.serverID = serverIDself.serverSocket = serverSocketself.clientsocket = Noneself.addr = Noneself.runflag = Trueself.connectList = ["connect","disconnect"]def run(self):print("Waiting for customer......")self.clientsocket,self.addr = self.serverSocket.accept()print("Customer IP: %s" % str(self.addr)+" is linking!")self.sendFlag(0)self.getMessage()def getMessage(self):while self.runflag:try:data = self.clientsocket.recv(1024).decode('utf-8')self.sendText(data)except Exception as reason:self.sendMessage(str(reason))self.sendText(str(self.addr)+" break connect...")self.sendFlag(1)breakself.clientsocket.close()print("Thread close successfully")def sendToClient(self,info):try:self.clientsocket.send(info.encode("utf-8"))print("boardcast success")except Exception as reason:print("boardcast failed:",reason)self.sendMessage(self.addr+" break connect...")self.sendFlag(1)def sendMessage(self,message):self._signal.emit("@@@".join([self.serverID,message]))def sendText(self,text):self._text.emit(text)def sendFlag(self,flagIndex):self._flag.emit("@@@".join([self.serverID,self.connectList[flagIndex]]))if __name__ == "__main__":app = QApplication(sys.argv)print("Here is Server! - input 'Server' to Create a Server")print("localhost IP: 82.157.140.78")ip = "0.0.0.0"hostName = "Server"port = 9995server = Server(ip,hostName,port)app.exec_()

6.大坑提醒

(1) 要想使用socket的TCP协议通过指定端口连接服务器得在服务器安全设置中开启指定端口才行，不然你怎么都连接不上

(2) 校园网对于网络专业来说就是个大坑，通常学校为了保证网络安全会做内网隔断,即内网内的主机之间不能相互通信，如果你ping不通对方主机ip就尝试用手机流量开热点吧...

(3) ubuntu上第一次运行PyQt5或PySide2可能会丢出"This application failed to start because no Qt platform plugin could be initialized"之类的错误，主要原因就是环境没配好(配了我一个下午T_T),看这篇文章可能会解决你的问题>>>解决方案

7.结束语

本来做好了本地连接测试后就直接拿到老师那里去给他看，谁知道老师看完之后直接当着全班的面说我这项目做的还有待欠缺，咱们学网络的不能局限于本地，要通过实际的网络连接测试才算是真正的完成了这个项目。说完全班都知道我内卷了....

后来自己又继续使用租来的服务器完成的跨网段通信连接测试，连接成功后开心的不得了!
本文的代码希望读者以及后来的学弟学妹们找来不是完全的照搬，只有真正自己实践过了，真正明白了，才算是真正的学到了知识!(要是课设被老师在网上找到一样的可是会直接不及格的哦~（笑）)