基于UDP的多人在线聊天室

Keywords: #技术 #Socket #Python #Flet
Table of Contents

实验内容

编写基于UDP的聊天室程序，实现多人聊天功能。自己设计应用协议，要求实现以下功能：

  1. 用户注册：服务器端记录已注册的用户名和密码。
  2. 用户登录：服务器接收登录信息后，检测是否在注册名单内。是，则聊天室的“在线清单”内加入此用户（包含用户名和主机连接地址）；否，则提示用户需要先注册。
  3. 公聊：客户端输入公聊命令+发送信息；服务器端接收信息，并按照指令，将信息发送往所有“在线清单”用户。
  4. 私聊：客户端输入私聊命令+发送方用户名+发送信息；服务器端接收信息，并按照指令，将信息发送往客户端指定的发送方用户。
  5. 退出聊天室。

程序设计思路

网络结构拓扑图

一对多的网络结构，即一台服务器对应多台客户端的星型拓扑结构

网络结构拓扑图

使用的Python库

  • socket套接字库
  • threading多线程库
  • flet第三方库，用于实现图形化界面

具体设计流程

首先我们应该先知道，UDP和TCP的区别是什么。UDP是无连接的，发送出去的数据不论对方是否收到，都不管，而TCP是面向连接的，每次通信都需要确保对方已经接收到才能进行下一步，也就多了很多协议设计的难度以及需要很多的异常判断。

在此之前，我已经写过一个关于socket的项目，是基于TCP的：Python远程控制程序，相比较之前这个项目，此次的程序设计就显得简单多了，毕竟是基于UDP的。

最简单的UDP连接

首先先写一个最简单的UDP连接程序，作为整个程序的基础

client端

IP = 'localhost'
POST = 7777
SERVER_ADDR = (IP, POST)
# 创建一个UDP的socket
client = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
# 发送消息
message = input()
client.sendto(message.encode(), SERVER_ADDR)
# 接收消息
# while True:
    # data, server = client.recvfrom(1024)
    # print(data.decode())

server端

IP = 'localhost'
POST = 7777
SERVER_ADDR = (IP, POST)
# 创建一个UDP的socket
server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
# 监听本机端口
server.bind(SERVER_ADDR)
# 接收消息
while True:
    data, address = server.recvfrom(BUFFER)
    message = data.decode()
    print(message)

这样就完成一个最简单的UDP连接程序，实现了client发送消息，server接收消息

相应功能的实现

接下里就是在上面程序的基础上加上对应的功能，本次程序需要实现：

  1. 登录
  2. 注册
  3. 公聊
  4. 私聊
  5. 退出

在client端中，我们需要将用户输入的消息，根据相应的操作进行封装成固定的格式，以便server端根据收到的数据，了解需要进行什么操作，经过操作后返回对应的数据

  • 注册操作的封装格式：REGISTER {username} {password}
  • 登录操作的封装格式：LOGIN {username} {password}
  • 公聊操作的封装格式：PUBLIC {message}
  • 私聊操作的封装格式：PRIVATE {to_username} {message}
  • 退出操作的封装格式：EXIT {username}

client端功能实现

# 注册
username = input("[*] 输入用户名: ")
password = input("[*] 输入密码: ")
registration_message = f"REGISTER {username} {password}"
client.sendto(registration_message.encode(), SERVER_ADDR)
response, add = client.recvfrom(BUFFER)
print(response.decode())

# 登录
username = input("[*] 输入用户名: ")
password = input("[*] 输入密码: ")
login_message = f"LOGIN {username} {password}"
client.sendto(login_message.encode(), SERVER_ADDR)
response, add = client.recvfrom(1024)
if response.decode() == "Login successful":
	# print(f"[+] 欢迎 {username} 加入聊天室")
	rev, add = client.recvfrom(1024)
	online_users = list(rev.decode().replace("[", "").replace("]", "").replace("'", "").split(", "))
	print("[*] 当前在线用户: ", end='')
	for user in online_users:
		print(user, end=' ')
	break
else:
	print(response.decode())

# 公聊、私聊、退出
message = input()
client.sendto(message.encode(), SERVER_ADDR)
if message == "exit":
	break
elif message.startswith("@"):
	to_address = message.split(" ")[0].replace("@", "")
	private_message = message.split(" ")[1]
	private_message = f"PRIVATE {to_address} {private_message}"
	client.sendto(private_message.encode(), SERVER_ADDR)
else:
	public_message = f"PUBLIC {message}"
	client.sendto(public_message.encode(), SERVER_ADDR)

server端功能实现

# 记录注册过的用户信息字典
user_dict = {}  # {username: password}
# 记录在线用户信息字典
online_user = {}  # {username: (ip, port)}

# 初始化用户信息
def init_users():
    with open('users.txt', 'r') as f:
        for line in f.readlines():
            username = line.split(',')[0]
            password = line.split(',')[1].replace('\n', '')
            user_dict[username] = password
    print(f'[+] 注册用户信息加载成功')

# 保存注册用户信息
def save_users():
    with open('users.txt', 'w') as f:
        for username in user_dict.keys():
            password = user_dict[username]
            f.write(f'{username},{password}\n')

# 列出在线用户
def list_online_users(address):
    users = []
    for username in online_user.keys():
        users.append(username)
    server.sendto(str(users).encode(), address)

# 注册
def register(username, password, address):
    if username in user_dict.keys():
        server.sendto("[-] 用户名已存在".encode(), address)
    else:
        user_dict[username] = password
        save_users()
        server.sendto("[+] 注册成功".encode(), address)

# 登录
def login(username, password, address):
    if username not in user_dict.keys():
        server.sendto("[-] 用户名不存在".encode(), address)
    elif user_dict[username] != password:
        server.sendto("[-] 密码错误".encode(), address)
    else:
        server.sendto("Login successful".encode(), address)
        online_user[username] = address
        # 发送在线用户列表
        list_online_users(address)
        # 发送上线通知
        for online in online_user.keys():
            server.sendto(f'\n[+] 欢迎 {username} 加入聊天室'.encode(), online_user[online])

# 公聊
def public_chat(message, address):
    from_username = ""
    # 通过address获取username
    for username in online_user.keys():
        if online_user[username] == address:
            from_username = username
            break
    for username in online_user.keys():
        server.sendto(f'[+] {from_username}: {message}'.encode(), online_user[username])

# 私聊
def private_chat(message, to_username, address):
    from_username = ""
    # 通过address获取username
    for username in online_user.keys():
        if online_user[username] == address:
            from_username = username
            break
    if to_username not in online_user.keys():
        server.sendto("[-] 用户不在线".encode(), address)
    else:
        to_address = online_user[to_username]
        server.sendto(f'[@] {from_username}: {message}'.encode(), to_address)
        server.sendto(f'[@] {from_username}: {message}'.encode(), address)

# 根据选项分别进入对应的函数
def menu(data, address):
    command = data.split()[0]
    if command == "REGISTER":
        username = data.split()[1]
        password = data.split()[2]
        register(username, password, address)
    elif command == "LOGIN":
        username = data.split()[1]
        password = data.split()[2]
        login(username, password, address)
    elif command == "PUBLIC":
        message = data.split()[1]
        public_chat(message, address)
    elif command == "PRIVATE":
        to_username = data.split()[1]
        message = data.split()[2]
        private_chat(message, to_username, address)

写完这些，其实整个实验就算是完成了，但是在测试的过程中还是发现了一些问题：用户输入信息的交互不好，因为是在命令行中进行操作，所以有些时候会和对方输出的信息重叠，受到干扰，体验不好，所以决定在接下来要写个图形化界面

图形化界面编写

对于python图形化界面的编写，其实有很多的方式，如python的标准库Tkinter、QT系列的pyqt5和pyside6、还有Flutter系的flet、…………

在此之前，我也利用flet编写过python的图形化界面项目：基于Python的本地量子文件加密，所以这次我还是使用flet进行编写

不管是之前的项目，还是这次的程序，编写图形化界面，我全部都是参考官方的文档，写的很详细，给予了我很大的帮助，里面还有很多的例子，帮助我理解，所以这次的编写也算是有点入门了

登录页面代码框架：

def login_page(page:ft.page):
	# 初始化设置
	page.title = "多人在线聊天室"  
	page.window_bgcolor = ft.colors.TRANSPARENT  
	page.bgcolor = "#E1F7FB"
	page.window_width = 500  
	page.window_height = 360  
	page.window_min_width = 500  
	page.window_min_height = 360
	# ……………………
	# 登录按钮点击事件  
	def btn_login_click(e):  
	    pass
	    
	# 注册按钮点击事件  
	def btn_register_click(e):  
	    pass
	    
	# 错误弹窗
	login_error_dlg = ft.AlertDialog(  
    title=ft.Text(  
        "用户名或密码错误！\n\n请检查后重新输入！"),  
	)  
	login_success_dlg = ft.AlertDialog(  
	    title=ft.Text(  
	        "登录成功！\n\n欢迎进入聊天室！"),  
	)  
	register_error_dlg = ft.AlertDialog(  
	    title=ft.Text(  
	        "注册失败！\n\n用户名已存在！"),  
	)  
	register_success_dlg = ft.AlertDialog(  
	    title=ft.Text(  
	        "注册成功！\n\n请登录！"),  
	)  
	login_empty_dlg = ft.AlertDialog(  
	    title=ft.Text(  
	        "用户名或密码为空！\n\n请检查后重新输入！")  
	)  
	server_error_dlg = ft.AlertDialog(  
	    title=ft.Text(  
	        "服务器错误！\n\n请稍后尝试！")  
	)
	# 标题
	title = ft.Stack(
	# ………………
	)
	# 密码输入框
	password_box = ft.TextField(  
    label="密码",  
    hint_text="请输入你的密码",  
    max_lines=1,  
    width=350,  
    height=55,  
    password=True,  
    can_reveal_password=True,  
    keyboard_type=ft.KeyboardType.VISIBLE_PASSWORD,  
    shift_enter=True,  
    on_submit=btn_login_click  
	)  
	# 用户名输入框
	username_box = ft.TextField(  
	    label="用户名",  
	    hint_text="请输入你的用户名",  
	    max_lines=1,  
	    width=350,  
	    height=55,  
	    autofocus=True,  
	    shift_enter=True,  
	    on_submit=lambda e: password_box.focus(),  
	)  
	# 登录按钮 
	btn_login = ft.ElevatedButton(  
	    text="登录",  
	    width=150,  
	    height=50,  
	    animate_size=True,  
	    on_click=btn_login_click,  
	    icon=ft.icons.LOGIN,  
	)  
	# 注册按钮
	btn_register = ft.ElevatedButton(  
	    text="注册",  
	    width=150,  
	    height=50,  
	    animate_size=True,  
	    on_click=btn_register_click,  
	    icon=ft.icons.CREATE_OUTLINED,  
	)
	# 添加页面布局
	page.add(
	# ……………………
	)

主页面代码框架：

def main_page(page: ft.Page):
	# 页面配置初始化
	page.clean()  
	# page.vertical_alignment = ft.MainAxisAlignment.START  # 垂直居中  
	page.horizontal_alignment = "stretch"  
	page.window_max_height = 600  
	page.window_max_width = 800  
	page.window_width = 800  
	page.window_height = 600  
	# 拦截本机窗口关闭信号 配合关闭窗口默认退出聊天  
	page.window_prevent_close = True

	# 接收消息
	def receive_message():
		pass
		
	# 更新聊天框  
	def update_chat(message):
		pass
	# 聊天内容列表  
	chat = ft.ListView(  
	    width=400,  
	    height=400,  
	    expand=True,  
	    spacing=10,  
	    auto_scroll=True,  
	)
	# 发送消息事件  
	def send_message_click(e):
		pass
		
	# 发送消息按钮  
	btn_send = ft.ElevatedButton(  
	    text="发送",  
	    width=120,  
	    height=50,  
	    animate_size=True,  
	    on_click=send_message_click,  
	    icon=ft.icons.SEND,  
	)

	# 输入框  
	new_message = ft.TextField(  
	    label="Message",  
	    hint_text="输入聊天消息",  
	    max_lines=1,  
	    width=500,  
	    height=60,  
	    autofocus=True,  
	    shift_enter=True,  
	    on_submit=send_message_click,  
	)

	# 当前用户头像  
	user_avatar = ft.CircleAvatar(  
	    content=ft.Text(ct.now_username, size=14, weight=ft.FontWeight.BOLD),  
	    width=60,  
	    height=60,  
	)

	# 添加页面布局
	page.add(
	# ………………
	)

	# 窗口关闭默认退出聊天室  
    def close_window(e):  
        if e.data == 'close':  
            ct.send_message('exit')  
            page.window_destroy() # 真正退出窗口  
  
    page.on_window_event = close_window  
    # 开启接收消息线程  
    receive_thread = threading.Thread(target=receive_message)  
    receive_thread.start()  
  
  
ft.app(target=login_page, assets_dir="assets")

成果展示

服务端启动 登录界面 注册功能 聊天界面 公聊功能 私聊功能 退出聊天室功能

总结与心得

总的来说，写一个基于UDP协议的多人聊天室这个实验还是很简单的，因为UDP是无连接的，不像TCP一样要考虑很多的问题，如丢包、失去连接等，UDP只管发送和接收，不用关心信息是否送达，所以少了很多“握手”的过程，简单不少。

但是，等到整个程序写完之后，在命令行中执行，我发现了一个问题，当我正在命令行输入消息时，终端上已经有部分文字，这时候对方发来消息，那么对方的消息就会从我当前的光标处打印出来，把我的语句断开，这对用户来说是个不好的体验。所以最后，我决定写一个图形化界面，将用户输入和公屏输出分开，这样体验和美观程度会大大提高。

图像化界面选择了 flet框架，它基于Google 的 Flutter实现，所以对我来说很快能上手，而不像pyqt5那样复杂。在编写图像化的过程中也遇到了很多问题，不过大部分是flet框架的问题，对于socket这一块的问题就是：因为UDP基于无连接，所以我发送出去的消息不知道对方有没有收到，当我的服务端没有开启时，客户端在登录时会向服务端发送请求同时转到阻塞等到回应，但是这个过程中没有提示，用户可能会多次点击登录或者注册，导致开启大量的UDP阻塞等待，等到服务端上线时，发送的消息会被这些阻塞等待给覆盖掉，导致影响后面的欢迎界面的输出。解决办法就是：设置一个阻塞等待超时，当等待超时，自动释放阻塞的线程，同时弹出警告框，告诉用户“当前服务器连接不上，稍后再试”，以免开启更多的线程，浪费资源同时影响后续输出。

解决了上面这个问题，但是上面这个解决办法同样导致出现了接下来一个问题：需要一直阻塞等待服务端发送消息的功能，因为上面设置的超时时间而报错，所以解决办法也很简单就是在这个功能中，将超时时间设为None。

总之，在这个实验中，我遇到了不少问题，随着一个个问题被解决，我也从中学到了很多。

← prev next →