【下载文档: python实现简单聊天室功能 可以私聊.txt 】
python实现简单聊天室功能 可以私聊
本文实例为大家分享了python实现简单聊天室功能的具体代码，供大家参考，具体内容如下
公共模块
首先写一个公共类,用字典的形式对数据的收发,并且进行封装,导入struct解决了TCP的粘包问题,并在公共类中进行了异常处理
import socket,struct,json
def send_dic(c,dic):
dic_json=json.dumps(dic)
dic_json_length=len(dic_json.encode('utf-8'))
struct_dic_json_length=struct.pack('q',dic_json_length)
c.send(struct_dic_json_length)
c.send(dic_json.encode('utf-8'))
def get_dic(c):
try:
dic_length=struct.unpack('q',c.recv(8))[0]
except:
return {'msg':'exit'}
try:
dic_json=c.recv(dic_length).decode('utf-8')
except:
return {'msg':'exit'}
dic=json.loads(dic_json)
return dic
服务器端
import socket
from concurrent.futures import ThreadPoolExecutor
import lib.common #导入写在lib里面的公共模块,代码在上面
import re
#进行开启服务器等一系列操作
s=socket.socket()
ip_host=('127.0.0.1',8000)
s.bind(ip_host)
s.listen()
#创建一个列表,用来保存客户端及其信息
c_list=[]
def get_send_msg(c,addr,c_list):
while True:
tag=False
dic=lib.common.get_dic(c)
if dic['msg']=='exit':
#如果接受出异常,或是客户端主动输入为exit,在列表中移除客户端信息
for i in c_list:
if i['addr']==addr:
c_list.remove(i)
break
if dic['is_siliao']==True:
#客户端发来的字典里面如果is_siliao==True,进入私聊代码
for i in c_list:
#遍历列表,并用正则表达式截取信息
li=re.findall('(.*?)@%s(.*)'%i['name'],dic['msg'])
if len(li)!=0:
dic['msg']=li[0][0]+li[0][1]
lib.common.send_dic(i['client'],dic)
tag=True
break
if tag:
continue
#如果不是私聊,进入下面代码,在聊天室进行群聊
for i in c_list:
if i['addr']!=addr:
lib.common.send_dic(i['client'],dic)
while True:
#用线程池,进行多次连接
print('客户端等待连接')
c,addr=s.accept()
print('%s连接了服务器'%addr[1])
name=c.recv(1024).decode('utf-8')#进行第一次接受,接受客户端的名字,为私聊的功能做准备
c_dic={'addr':addr,'client':c,'name':name}#将客户端的信息保存在字典中
c_list.append(c_dic)#将字典加入列表
t=ThreadPoolExecutor()
t.submit(get_send_msg,c,addr,c_list)
客户端:
import lib.common
from concurrent.futures import ThreadPoolExecutor
c=socket.socket()
ip_host=('127.0.0.1',8000)
c.connect(ip_host)
def send_msg(c,name):
while True:
msg = input ('>>:').strip ()
is_siliao=False
if not msg:
continue
# if msg.startswith('@'):
if '@'in msg:
is_siliao=True
dic = {'msg': msg,'name':name,'is_siliao':is_siliao}
lib.common.send_dic(c,dic)
if msg=='exit':
c.close ()
break
def get_msg(c):
while True:
dic=lib.common.get_dic(c)
if dic['is_siliao']==True:
print('来自%s的私聊:%s'%(dic['name'],dic['msg']))
continue
print('%s:%s'%(dic['name'],dic['msg']))
t=ThreadPoolExecutor()
name=input('你的聊天名字:').strip()
c.send(name.encode('utf-8'))
t.submit(send_msg,c,name)
t.submit(get_msg,c)
运行代码截图:
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持中文源码网。
