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)

실행 코드 캡 처:





이상 이 바로 본 고의 모든 내용 입 니 다.여러분 의 학습 에 도움 이 되 고 저 희 를 많이 응원 해 주 셨 으 면 좋 겠 습 니 다.