本文實例為大家分享了python實現(xiàn)簡單聊天室功能的具體代碼，供大家參考，具體內容如下
公共模塊
首先寫一個公共類,用字典的形式對數(shù)據(jù)的收發(fā),并且進行封裝,導入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()
#創(chuàng)建一個列表,用來保存客戶端及其信息
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:
#客戶端發(fā)來的字典里面如果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)
運行代碼截圖:
以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持腳本之家。
更多文章、技術交流、商務合作、聯(lián)系博主
微信掃碼或搜索：z360901061
微信掃一掃加我為好友
QQ號聯(lián)系： 360901061
您的支持是博主寫作最大的動力，如果您喜歡我的文章，感覺我的文章對您有幫助，請用微信掃描下面二維碼支持博主2元、5元、10元、20元等您想捐的金額吧，狠狠點擊下面給點支持吧，站長非常感激您！手機微信長按不能支付解決辦法：請將微信支付二維碼保存到相冊，切換到微信，然后點擊微信右上角掃一掃功能，選擇支付二維碼完成支付。
【本文對您有幫助就好】元
