当前位置： 首页 > news >正文

实验2 python的TCP群聊系统实现

news 来源：原创 2025/4/24 8:03:16

实验2 TCP群聊系统实现

一、实验目的和任务

  1. 掌握TCP Socket编程的基本原理与实现方法
  2. 理解多线程在网络编程中的应用
  3. 实现基于TCP协议的群聊系统
  4. 学习GUI与网络编程的结合应用

二、实验内容

1.服务器端程序开发：实现用户连接管理、消息广播功能、私聊消息处理、在线用户列表维护

2.客户端程序开发：用户界面设计、消息发送与接收、私聊功能实现、在线用户列表显示

3.系统测试：多客户端连接测试、消息收发测试、私聊功能测试

三、实验步骤

实验2.1 服务器端实现

代码参考：

mport socket
import threading
import time
class ChatServer:def __init__(self, host='localhost', port=12345):self.host = hostself.port = portself.server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)self.clients = []self.nicknames = []def start(self):self.server.bind((self.host, self.port))self.server.listen()print(f"服务器已启动，监听 {self.host}:{self.port}")# 在start方法中启动用户列表广播线程userlist_thread = threading.Thread(target=self.broadcast_userlist)userlist_thread.daemon = Trueuserlist_thread.start()
持续接收用户连接：while True:client, address = self.server.accept()print(f"新连接来自: {str(address)}")# 要求客户端发送昵称client.send('NICK'.encode('utf-8'))nickname = client.recv(1024).decode('utf-8')self.nicknames.append(nickname)self.clients.append(client)print(f"昵称是: {nickname}")self.broadcast(f"{nickname} 加入了聊天室!".encode('utf-8'))# 为每个客户端启动线程thread = threading.Thread(target=self.handle_client, args=(client,))thread.start()
广播用户列表：def broadcast_userlist(self):while True:try:userlist = "USERLIST:" + ",".join(self.nicknames)self.broadcast(userlist.encode('utf-8'))time.sleep(5)  # 每5秒更新一次except:break
广播信息：def broadcast(self, message):for client in self.clients:try:client.send(message)except:# 移除断开连接的客户端index = self.clients.index(client)self.clients.remove(client)client.close()nickname = self.nicknames[index]self.broadcast(f"{nickname} 离开了聊天室!".encode('utf-8'))self.nicknames.remove(nickname)
处理客户线程的具体函数，负责接收消息：def handle_client(self, client):while True:try:message = client.recv(1024).decode('utf-8')# 处理私聊消息if message.startswith("/pm"):parts = message.split(" ", 2)  # 格式: /pm 目标昵称 消息内容if len(parts) == 3:target_nick = parts[1]if target_nick in self.nicknames:target_index = self.nicknames.index(target_nick)private_msg = f"[私聊] {self.nicknames[self.clients.index(client)]}: {parts[2]}"self.clients[target_index].send(private_msg.encode('utf-8'))# 给发送者回显client.send(f"[私聊] 你 -> {target_nick}: {parts[2]}".encode('utf-8'))continue  # 私聊消息不广播# 普通消息广播self.broadcast(message.encode('utf-8'))except:# 移除断开连接的客户端index = self.clients.index(client)self.clients.remove(client)client.close()nickname = self.nicknames[index]self.broadcast(f"{nickname} 离开了聊天室!".encode('utf-8'))self.nicknames.remove(nickname)break

完整python脚本如下:

import socket
import threading
import timeclass ChatServer:def __init__(self, host='localhost', port=12345):self.host = hostself.port = portself.server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)self.clients = []self.nicknames = []def start(self):self.server.bind((self.host, self.port))self.server.listen()# 获取本机IP地址hostname = socket.gethostname()ip_address = socket.gethostbyname(hostname)print(f"服务器已启动，监听 {ip_address}:{self.port}")print(f"本地地址: {self.host}:{self.port}")print("\n客户端连接说明:")print("1. 打开新的终端窗口")print("2. 运行命令: python 1220911101-fyt-2-client.py")print("3. 在客户端界面输入服务器地址: {ip_address}")print("4. 输入端口号: {self.port}")print("5. 输入您的昵称即可开始聊天")print("\n提示: 可以同时运行多个客户端程序进行测试")# 在start方法中启动用户列表广播线程userlist_thread = threading.Thread(target=self.broadcast_userlist)userlist_thread.daemon = Trueuserlist_thread.start()while True:client, address = self.server.accept()print(f"新连接来自: {str(address)}")# 要求客户端发送昵称client.send('NICK'.encode('utf-8'))nickname = client.recv(1024).decode('utf-8')self.nicknames.append(nickname)self.clients.append(client)print(f"昵称是: {nickname}")self.broadcast(f"{nickname} 加入了聊天室!".encode('utf-8'))# 为每个客户端启动线程thread = threading.Thread(target=self.handle_client, args=(client,))thread.start()def broadcast_userlist(self):while True:try:userlist = "USERLIST:" + ",".join(self.nicknames)self.broadcast(userlist.encode('utf-8'))time.sleep(5)  # 每5秒更新一次except:breakdef broadcast(self, message):for client in self.clients:try:client.send(message)except:# 移除断开连接的客户端index = self.clients.index(client)self.clients.remove(client)client.close()nickname = self.nicknames[index]self.broadcast(f"{nickname} 离开了聊天室!".encode('utf-8'))self.nicknames.remove(nickname)def handle_client(self, client):while True:try:message = client.recv(1024).decode('utf-8')# 处理私聊消息if message.startswith("/pm"):parts = message.split(" ", 2)  # 格式: /pm 目标昵称 消息内容if len(parts) == 3:target_nick = parts[1]if target_nick in self.nicknames:target_index = self.nicknames.index(target_nick)private_msg = f"[私聊] {self.nicknames[self.clients.index(client)]}: {parts[2]}"self.clients[target_index].send(private_msg.encode('utf-8'))# 给发送者回显client.send(f"[私聊] 你 -> {target_nick}: {parts[2]}".encode('utf-8'))continue  # 私聊消息不广播# 普通消息广播self.broadcast(message.encode('utf-8'))except:# 移除断开连接的客户端index = self.clients.index(client)self.clients.remove(client)client.close()nickname = self.nicknames[index]self.broadcast(f"{nickname} 离开了聊天室!".encode('utf-8'))self.nicknames.remove(nickname)breakif __name__ == "__main__":server = ChatServer()server.start()

功能效果如下图所示: 

实验2.2 客户端实现

核心代码参考

class ChatClient:def __init__(self, host='localhost', port=12345):# 初始化self.host = hostself.port = portself.client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)self.nickname = ""# 创建GUIself.root = Tk()self.root.title("群聊客户端")# 聊天显示区域self.chat_area = scrolledtext.ScrolledText(self.root, wrap=WORD, width=50, height=20)self.chat_area.pack(padx=10, pady=10)self.chat_area.config(state=DISABLED)# 消息输入区域self.msg_entry = Entry(self.root, width=40)self.msg_entry.pack(padx=10, pady=5)self.msg_entry.bind("<Return>", self.send_message)# 发送按钮self.send_btn = Button(self.root, text="发送", command=self.send_message)self.send_btn.pack(pady=5)# 在__init__方法中添加帮助标签self.help_label = Label(self.root, text="私聊格式: /pm 昵称 消息", fg="gray")self.help_label.pack()# 用户列表框架self.user_frame = Frame(self.root)self.user_frame.pack(side=RIGHT, padx=10, pady=10)self.user_label = Label(self.user_frame, text="在线用户")self.user_label.pack()self.user_list = Listbox(self.user_frame, width=20, height=15)self.user_list.pack()# 双击用户列表发起私聊self.user_list.bind("<Double-Button-1>", self.start_private_chat)
启动连接方法，在初始化完成后调用。def connect(self):try:self.client.connect((self.host, self.port))# 获取昵称self.nickname = simpledialog.askstring("昵称", "请输入你的昵称:")if not self.nickname:self.nickname = "匿名用户"self.client.send(self.nickname.encode('utf-8'))# 启动接收消息的线程receive_thread = threading.Thread(target=self.receive)receive_thread.daemon = Truereceive_thread.start()self.root.mainloop()except Exception as e:messagebox.showerror("错误", f"无法连接到服务器: {e}")self.root.destroy()
接收消息的具体函数：def receive(self):while True:try:message = self.client.recv(1024).decode('utf-8')if message.startswith("USERLIST:"):# 更新用户列表users = message[len("USERLIST:"):].split(",")self.user_list.delete(0, END)for user in users:if user and user != self.nickname:  # 不显示自己self.user_list.insert(END, user)else: #普通消息self.display_message(message)except:self.display_message("与服务器的连接已断开!")self.client.close()break
进行私聊的方法：def start_private_chat(self, event):selected = self.user_list.get(self.user_list.curselection())self.msg_entry.delete(0, END)self.msg_entry.insert(0, f"/pm {selected} ")self.msg_entry.focus()
在聊天框中显示信息：def display_message(self, message):self.chat_area.config(state=NORMAL)self.chat_area.insert(END, message + "\n")self.chat_area.config(state=DISABLED)self.chat_area.see(END)
向服务端发送消息：def send_message(self, event=None):message = self.msg_entry.get()if message:# 检查是否是私聊命令if message.startswith("/pm"):# 直接发送原始命令到服务器full_message = messageelse:full_message = f"{self.nickname}: {message}"try:self.client.send(full_message.encode('utf-8'))self.msg_entry.delete(0, END)except:self.display_message("发送失败，请检查连接!")

完整python脚本如下:

import socket
import threading
from tkinter import *
from tkinter import scrolledtext, messagebox, simpledialogclass ChatClient:def __init__(self, host='localhost', port=12345):# 初始化self.host = hostself.port = portself.client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)self.nickname = ""# 创建GUIself.root = Tk()self.root.title("TCP群聊客户端")self.root.geometry("800x600")self.root.minsize(600, 400)# 主框架main_frame = Frame(self.root)main_frame.pack(fill=BOTH, expand=True, padx=10, pady=10)# 用户列表框架user_frame = Frame(main_frame, width=150, relief=RAISED, borderwidth=1)user_frame.pack(side=RIGHT, fill=Y, padx=(10,0))Label(user_frame, text="在线用户", font=('Arial', 10, 'bold')).pack(pady=5)self.user_list = Listbox(user_frame, selectmode=SINGLE)self.user_list.pack(fill=BOTH, expand=True, padx=5, pady=5)self.user_list.bind("<Double-Button-1>", self.start_private_chat)# 聊天区域框架chat_frame = Frame(main_frame)chat_frame.pack(side=LEFT, fill=BOTH, expand=True)# 聊天显示区域self.chat_area = scrolledtext.ScrolledText(chat_frame, wrap=WORD, font=('Arial', 10), padx=5, pady=5)self.chat_area.pack(fill=BOTH, expand=True)self.chat_area.config(state=DISABLED)# 输入区域框架input_frame = Frame(chat_frame)input_frame.pack(fill=X, pady=(10,0))# 消息输入区域self.msg_entry = Entry(input_frame, font=('Arial', 10),relief=SOLID)self.msg_entry.pack(side=LEFT, fill=X, expand=True, padx=(0,5))self.msg_entry.bind("<Return>", self.send_message)# 发送按钮self.send_btn = Button(input_frame, text="发送", command=self.send_message,width=8,relief=RAISED)self.send_btn.pack(side=RIGHT)# 帮助标签help_frame = Frame(chat_frame)help_frame.pack(fill=X, pady=(5,0))self.help_label = Label(help_frame, text="提示: 双击用户列表可发起私聊 | 私聊格式: /pm 昵称 消息", fg="gray",font=('Arial', 8))self.help_label.pack(side=LEFT)# 双击用户列表发起私聊self.user_list.bind("<Double-Button-1>", self.start_private_chat)def connect(self):try:# 获取服务器地址和端口server_ip = simpledialog.askstring("服务器地址", "请输入服务器IP地址(如:192.168.1.100):", initialvalue=self.host)if not server_ip:server_ip = self.hostport_str = simpledialog.askstring("端口号", "请输入服务器端口号(默认12345):", initialvalue=str(self.port))try:port = int(port_str) if port_str else self.portexcept ValueError:messagebox.showwarning("警告", "端口号必须是数字，将使用默认端口12345")port = self.port# 测试连接self.client.settimeout(3)  # 设置3秒超时self.client.connect((server_ip, port))self.client.settimeout(None)  # 连接成功后取消超时# 获取昵称self.nickname = simpledialog.askstring("昵称", "请输入你的昵称:")if not self.nickname:self.nickname = "匿名用户"self.client.send(self.nickname.encode('utf-8'))# 启动接收消息的线程receive_thread = threading.Thread(target=self.receive)receive_thread.daemon = Truereceive_thread.start()self.root.mainloop()except Exception as e:messagebox.showerror("错误", f"无法连接到服务器: {e}")self.root.destroy()def receive(self):while True:try:message = self.client.recv(1024).decode('utf-8')if message.startswith("USERLIST:"):# 更新用户列表users = message[len("USERLIST:"):].split(",")self.user_list.delete(0, END)for user in users:if user and user != self.nickname:  # 不显示自己self.user_list.insert(END, user)else: #普通消息self.display_message(message)except:self.display_message("与服务器的连接已断开!")self.client.close()breakdef start_private_chat(self, event):selected = self.user_list.get(self.user_list.curselection())self.msg_entry.delete(0, END)self.msg_entry.insert(0, f"/pm {selected} ")self.msg_entry.focus()def display_message(self, message):self.chat_area.config(state=NORMAL)self.chat_area.insert(END, message + "\n")self.chat_area.config(state=DISABLED)self.chat_area.see(END)def send_message(self, event=None):message = self.msg_entry.get()if message:# 检查是否是私聊命令if message.startswith("/pm"):# 直接发送原始命令到服务器full_message = messageelse:full_message = f"{self.nickname}: {message}"try:self.client.send(full_message.encode('utf-8'))self.msg_entry.delete(0, END)except:self.display_message("发送失败，请检查连接!")if __name__ == "__main__":client = ChatClient()client.connect()

功能效果如下图所示: 

四、思考题

1.为什么服务器需要为每个客户端创建单独的线程？如果不使用多线程会有什么问题？

答:原因：服务器需要同时处理多个客户端的请求，每个客户端的操作（如发送消息、接收消息）可能是独立的，且可能阻塞（如等待输入）。多线程允许服务器并行处理多个客户端的请求，提高并发性和响应速度。

 不使用多线程的问题：单线程模式下，服务器只能依次处理客户端的请求。如果一个客户端阻塞（如长时间未发送消息），其他客户端会被阻塞，导致系统无法及时响应。用户体验差，系统吞吐量低。

2.实验中如何处理客户端异常断开连接的情况？

答:处理方法：

心跳机制：定期检测客户端是否存活（如定时发送心跳包，超时未响应则判定为断开）。

异常捕获：在服务器线程中捕获客户端连接异常（如 SocketException），关闭对应的套接字和线程。

资源清理：从用户列表中移除断开连接的客户端，并通知其他用户更新列表。

3.私聊功能是如何实现的？服务器如何区分私聊消息和普通消息？

答:实现方式：客户端发送私聊消息时，需指定目标用户（如格式为 @username:message）。
服务器解析消息内容，检测是否有私聊标识（如 @username）。
如果是私聊消息，服务器仅将消息转发给目标用户；否则广播给所有用户。
区分机制：通过消息格式或协议字段（如消息头中包含 type: private 或 target: username）区分。

4.用户列表更新采用了什么机制？为什么需要单独线程处理？

答:机制：当用户加入或离开时，服务器更新用户列表，并通过广播通知所有客户端。
客户端接收更新后，刷新本地用户列表显示。
单独线程的原因：用户列表更新可能频繁且独立于消息收发，单独线程可以避免阻塞主线程或其他客户端线程。
提高系统响应速度和稳定性。

5.如果要将本系统改为UDP实现，哪些部分需要修改？会遇到什么挑战？

答:协议设计：UDP是无连接的，需实现自定义的可靠性机制（如消息序号、确认重传）。
消息格式：需添加字段标识消息类型、序号等。
客户端管理：UDP无连接状态，需维护客户端状态表（如IP和端口）。
挑战：
可靠性：UDP不保证交付，需处理丢包、乱序问题。
状态管理：需额外维护客户端连接状态（如超时检测）

6.本实验中的GUI界面使用了哪些组件？各自的作用是什么？

答:主要组件：
消息显示区（文本框）：显示聊天消息。
消息输入框：用户输入消息内容。
发送按钮：触发消息发送操作。
用户列表：显示当前在线用户。
私聊选择框（可选）：选择私聊目标用户。
作用：提供用户交互界面，支持消息收发、用户列表更新等功能。

7.如何改进当前系统以实现消息历史记录功能？

答:改进方案：
服务器端存储：将消息保存到数据库（如SQLite、MySQL）或文件（如日志文件）。
按时间或会话分类存储。
客户端加载：客户端启动时，从服务器请求历史消息。支持按时间范围或关键词检索。
优化：采用分页加载，避免一次性加载过多数据。支持消息加密存储（如敏感信息）。

相关文章：

  • Java基础：认识注解，模拟junit框架
  • 3dmax模型怎么导入酷家乐插件，材质贴图在，可优化不卡，可批量处理，无需打开一个个max，可批量转FBX/GLB/GLTF/OBJ/SU
  • 查询Hologres或postgresql中的数据
  • 即插即用模块(3) -LSK 特征提取
  • Spring--统一数据返回格式与统一异常处理
  • 将 MySQL 8 主从复制延迟优化到极致
  • 如何在iStoreOS DHCP中排除特定IP地址
  • zkPass案例实战之合约篇
  • 【论文#目标检测】Attention Is All You Need
  • 【泊松过程和指数分布】
  • 力扣DAY63-67 | 热100 | 二分：搜索插入位置、搜索二维矩阵、排序数组查找元素、搜索旋转排序数组、搜索最小值
  • OpenCV 图形API（52）颜色空间转换-----将 NV12 格式的图像数据转换为 RGB 格式的图像
  • 计算机视觉基础
  • 提高Spring Boot开发效率的实践
  • MsQuick编译和使用
  • c++概念——模板的进阶讲解
  • django软件开发招聘数据分析与可视化系统设计与实现(源码+lw+部署文档+讲解)，源码可白嫖!
  • 香港科技大学广州｜金融科技学域博士招生宣讲会—南开大学专场
  • ThinkPHP快速使用手册
  • VUE的创建
  • 继加州后，美国又有11州起诉特朗普政府滥用关税政策“违法”
  • 中国围棋协会将不组队参加今年的LG杯世界棋王赛
  • 证券时报：金价再创历史新高，“避险”黄金不应异化为投机工具
  • 广西一季度GDP为6833.92亿元，同比增长5.8%
  • 哈萨克斯坦一名副市长遭枪击
  • 深一度｜中国花样滑冰因何大滑坡