Python简易服务器

石泊远 · 发表于 2020-11-23 22:50:17

马上注册，结交更多好友，享用更多功能^_^

您需要登录才可以下载或查看，没有账号？立即注册

x

创造两个文件，一个是聊天室服务器，另外一个是聊天室客户端
聊天室服务器代码为：

import wx
from socket import *
import threading
import time
class StoneServer(wx.Frame):
def __init__(self): #c_name 客户端名字
#调用父类的初始化函数
wx.Frame.__init__(self,None,id = 100,title = '石泊远的服务器',pos = wx.DefaultPosition,size = (400,470))
pa = wx.Panel(self) #在窗口中初始化一个面板
#在面板里面会放一些按钮，文本框，文本输入框等，把这些对象统一放入一个盒子里面
box = wx.BoxSizer(wx.VERTICAL) #在盒子里面垂直方向自动排版
ga = wx.FlexGridSizer(wx.HORIZONTAL) #可升缩的网格布局,水平方向
#创建两个按钮
start_server_button = wx.Button(pa,size = (133,40),label = '启动')
record_save_button = wx.Button(pa,size = (133,40),label = '聊天记录保存')
stop_server_button = wx.Button(pa,size = (133,40),label = '停止')
ga.Add(start_server_button,1,wx.TOP)
ga.Add(record_save_button,1,wx.TOP)
ga.Add(stop_server_button,1,wx.TOP)
box.Add(ga,1,wx.ALIGN_CENTER)
#创造只读的文本框
self.text = wx.TextCtrl(pa,size=(400,400),style = wx.TE_MULTILINE | wx.TE_READONLY)
box.Add(self.text,1,wx.ALIGN_CENTER)
pa.SetSizer(box)
'''以上代码窗口结束'''
'''服务器准备执行的属性'''
self.isOn = False #服务器是否启动
self.host_port = ('',8888)
self.server_socket = socket(AF_INET,SOCK_STREAM)
self.server_socket.bind(self.host_port)
self.server_socket.listen(5)
self.session_thread_map = {}
'''给所有的按钮绑定相应的动作'''
self.Bind(wx.EVT_BUTTON,self.start_server,start_server_button)
self.Bind(wx.EVT_BUTTON,self.save_record,record_save_button)
#服务器开始启动函数
def start_server(self,event):
print('服务器开始')
if not self.isOn:
# 启动服务器的主线程
self.isOn = True
main_thread = threading.Thread(target = self.do_work,)
main_thread.setDaemon(True)
main_thread.start()
#服务器运行之后的函数
def do_work(self):
print('开始工作')
while self.isOn:
session_socket,client_addr = self.server_socket.accept()
#服务器首先接受客户端发来的消息
username = session_socket.recv(1024).decode('UTF-8')
#创建一个会话线程
session_thread = SessionThread(session_socket,username,self)
self.session_thread_map[username] = session_thread
session_thread.start()
self.show_info_and_send_client('服务器通知','欢迎%s进入聊天室！'%username,time.strftime('%Y-%m-%d %H:%M:%S',time.localtime()))
self.server_socket.close()
def show_info_and_send_client(self,source,data,data_time):
send_data = '%s : %s\n时间：%s\n' %(source,data,data_time)
self.text.AppendText('--------------------------------\n%s'%send_data)
for client in self.session_thread_map.values():
if client.isOn:
client.user_socket.send(send_data.encode('UTF-8'))
#服务器保存聊天记录
def save_record(self,event):
record = self.text.GetValue()
with open('record.log','w+')as f:
f.write(record)
#服务器端会话线程的类
class SessionThread(threading.Thread):
def __init__(self,socket,un,server):
threading.Thread.__init__(self)
self.user_socket = socket
self.username = un
self.server = server
self.isOn = True
def run(self):
print('客户端%s，已经和服务器连接成功，服务器启动一个会话线程'%self.username)
while self.isOn:
data = self.user_socket.recv(1024).decode('UTF-8')
if data == 'A^disconnect^B':
self.isOn = False
#有用户离开
self.server.show_info_and_send_client('服务器通知','%s离开聊天室！'%self.username,time.strftime('%Y-%m-%d %H:%M:%S',time.localtime()))
else:
# 其他聊天信息，我们应该显示给所有客户端，包括服务器
self.server.show_info_and_send_client(self.username,data,time.strftime('%Y-%m-%d %H:%M:%S',time.localtime()))
self.user_socket.close()
if __name__ == '__main__':
app = wx.App()
StoneServer().Show()
app.MainLoop() #循环刷新

复制代码

服务器客户端：

import wx
from socket import *
import threading
#客户端继承wx.Frame
class StoneClient(wx.Frame):
def __init__(self,c_name): #c_name 客户端名字
#调用父类的初始化函数
wx.Frame.__init__(self,None,id = 101,title = '%s的客户端'%c_name,pos = wx.DefaultPosition,size = (400,470))
pa = wx.Panel(self) #在窗口中初始化一个面板
#在面板里面会放一些按钮，文本框，文本输入框等，把这些对象统一放入一个盒子里面
box = wx.BoxSizer(wx.VERTICAL) #在盒子里面垂直方向自动排版
ga = wx.FlexGridSizer(wx.HORIZONTAL) #可升缩的网格布局,水平方向
#创建两个按钮
conn_button = wx.Button(pa,size=(200,40),label="连接")
dis_conn_button = wx.Button(pa,size=(200,40),label="离开")
ga.Add(conn_button,1,wx.TOP | wx.LEFT)
ga.Add(dis_conn_button,1,wx.TOP | wx.RIGHT)
box.Add(ga,1,wx.ALIGN_CENTER)
#创建聊天内容的文本框,不可以写消息
self.text = wx.TextCtrl(pa,size=(400,250),style = wx.TE_MULTILINE | wx.TE_READONLY)
box.Add(self.text,1,wx.ALIGN_CENTER)
self.input_text = wx.TextCtrl(pa,size=(400,100),style = wx.TE_MULTILINE)
box.Add(self.input_text,1,wx.ALIGN_CENTER)
g2 = wx.FlexGridSizer(wx.HORIZONTAL)
clear_button = wx.Button(pa,size=(200,40),label="重置")
send_button = wx.Button(pa,size=(200,40),label="发送")
g2.Add(clear_button,1,wx.TOP | wx.LEFT)
g2.Add(send_button,1,wx.TOP | wx.RIGHT)
box.Add(g2,1,wx.ALIGN_CENTER)
pa.SetSizer(box) #把盒子放入面板中
'''以上代码完成了客户端界面'''
'''给所有按钮绑定'''
self.Bind(wx.EVT_BUTTON,self.connect_to_server,conn_button)
self.Bind(wx.EVT_BUTTON,self.send_to,send_button)
self.Bind(wx.EVT_BUTTON,self.go_out,dis_conn_button)
self.Bind(wx.EVT_BUTTON,self.reset,clear_button)
'''客户端的属性'''
self.name = c_name
self.isConnected = False #客户端是否连上服务器
self.client_socket = None
#连接服务器
def connect_to_server(self,event):
print('客户端%s,开始连接服务器'%self.name)
if not self.isConnected:
server_host_port = ('localhost',8888)
self.client_socket = socket(AF_INET,SOCK_STREAM)
self.client_socket.connect(server_host_port)
#之前规定，客户端只要连接成功，马上把自己的名字发给服务器
self.client_socket.send(self.name.encode('UTF-8'))
t = threading.Thread(target = self.receive_data)
t.setDaemon(True)
self.isConnected = True
t.start()
#接受服务器发送过来的聊天数据
def receive_data(self):
print('服务器准备接受服务器的数据')
while self.isConnected:
data = self.client_socket.recv(1024).decode('UTF-8')
self.text.AppendText('%s\n'%data)
#客户端发送信息到客户端
def send_to(self,event):
if self.isConnected:
info = self.input_text.GetValue()
if info != '':
self.client_socket.send(info.encode('UTF-8'))
#输入框中的数据如果已经发送了，输入框重新为空
self.input_text.SetValue('')
#客户端离开聊天室
def go_out(self,event):
self.client_socket.send('A^disconnect^B'.encode('UTF-8'))
#客户端关闭
self.isConnected = False
#客户端信息重置
def reset(self,event):
self.input_text.Clear()
if __name__ == '__main__':
app = wx.App()
name = input('请输入客户端名字：')
StoneClient(name).Show()
app.MainLoop() #循环刷新

复制代码

账号		自动登录	找回密码
密码			立即注册

[作品展示] Python简易服务器

马上注册，结交更多好友，享用更多功能^_^

浏览过的版块