大橙子网站建设,新征程启航
为企业提供网站建设、域名注册、服务器等服务
这篇文章给大家分享的是有关使用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)
运行代码截图:
感谢各位的阅读!关于“使用python实现简单聊天室功能的案例”这篇文章就分享到这里了,希望以上内容可以对大家有一定的帮助,让大家可以学到更多知识,如果觉得文章不错,可以把它分享出去让更多的人看到吧!