python 간단 한 채 팅 애플 릿 구현
이것 은 python 을 사용 하여 간단 한 채 팅 방 을 실현 하 는 기능 입 니 다.그 안에 단체 채 팅,개인 채 팅 두 가지 채 팅 방식 이 포함 되 어 있 습 니 다.실현 하 는 방식 은 소켓 프로 그래 밍 을 사용 하 는 TCP 프로 토 콜 c/s 구 조 를 사용 하 는 채 팅 방 입 니 다.
사고의 방향 을 실현 하 다.
x01 서버 구축
우선,서버 에서 socket 을 사용 하여 메 시 지 를 받 아들 이 고 하나의 socket 요청 을 받 아들 일 때마다 새로운 스 레 드 를 열 어 메시지 의 배포 와 수용 을 관리 하 는 동시에 하나의 handler 가 존재 하여 모든 스 레 드 를 관리 하여 채 팅 방 의 각종 기능 에 대한 처 리 를 실현 합 니 다.
x02 클 라 이언 트 의 구축
클 라 이언 트 의 구축 은 서버 보다 훨씬 간단 합 니 다.클 라 이언 트 의 역할 은 메시지 의 발송 과 수용,그리고 특정한 규칙 에 따라 특정한 문 자 를 입력 하여 서로 다른 기능 의 사용 을 실현 하 는 것 입 니 다.따라서 클 라 이언 트 에 서 는 두 개의 라인 만 사용 해 야 합 니 다.하 나 는 메 시 지 를 받 아들 이 는 것 이 고 하 나 는 메 시 지 를 보 내 는 데 전문 적 인 것 입 니 다.
왜 하 나 를 사용 하지 않 는 지 에 대해 서 는 하나만 사용 하면 메 시 지 를 받 았 을 때 보 내기 전에 메 시 지 를 받 는 것 이 막 힌 상태 이기 때 문 입 니 다.마찬가지 로 메 시 지 를 보 내 는 것 도 마찬가지 입 니 다.그러면 이 두 가지 기능 을 한 곳 에 두 고 실현 하면 계속 보 내 거나 받 아들 일 수 없 기 때 문 입 니 다.
실현 방식
서버 구현
import json
import threading
from socket import *
from time import ctime
class PyChattingServer:
__socket = socket(AF_INET, SOCK_STREAM, 0)
__address = ('', 12231)
__buf = 1024
def __init__(self):
self.__socket.bind(self.__address)
self.__socket.listen(20)
self.__msg_handler = ChattingHandler()
def start_session(self):
print(' ...\r
')
try:
while True:
cs, caddr = self.__socket.accept()
# handler , socket
self.__msg_handler.start_thread(cs, caddr)
except socket.error:
pass
class ChattingThread(threading.Thread):
__buf = 1024
def __init__(self, cs, caddr, msg_handler):
super(ChattingThread, self).__init__()
self.__cs = cs
self.__caddr = caddr
self.__msg_handler = msg_handler
#
def run(self):
try:
print('... :', self.__caddr)
data = ' PY_CHATTING！ cooooool ( `)\r
'
self.__cs.sendall(bytes(data, 'utf-8'))
while True:
data = self.__cs.recv(self.__buf).decode('utf-8')
if not data:
break
self.__msg_handler.handle_msg(data, self.__cs)
print(data)
except socket.error as e:
print(e.args)
pass
finally:
self.__msg_handler.close_conn(self.__cs)
self.__cs.close()
class ChattingHandler:
__help_str = "[ SYSTEM ]\r
" \
" /ls, \r
" \
" /h, \r
" \
" @ ( )+ , \r
" \
" /i, \r
" \
" /i, \r
" \
" / "
__buf = 1024
__socket_list = []
__user_name_to_socket = {}
__socket_to_user_name = {}
__user_name_to_broadcast_state = {}
def start_thread(self, cs, caddr):
self.__socket_list.append(cs)
chat_thread = ChattingThread(cs, caddr, self)
chat_thread.start()
def close_conn(self, cs):
if cs not in self.__socket_list:
return
# socket
nickname = "SOMEONE"
if cs in self.__socket_list:
self.__socket_list.remove(cs)
# socket username
if cs in self.__socket_to_user_name:
nickname = self.__socket_to_user_name[cs]
self.__user_name_to_socket.pop(self.__socket_to_user_name[cs])
self.__socket_to_user_name.pop(cs)
self.__user_name_to_broadcast_state.pop(nickname)
nickname += " "
#
self.broadcast_system_msg(nickname + " PY_CHATTING")
#
def handle_msg(self, msg, cs):
js = json.loads(msg)
if js['type'] == "login":
if js['msg'] not in self.__user_name_to_socket:
if ' ' in js['msg']:
self.send_to(json.dumps({
'type': 'login',
'success': False,
'msg': ' '
}), cs)
else:
self.__user_name_to_socket[js['msg']] = cs
self.__socket_to_user_name[cs] = js['msg']
self.__user_name_to_broadcast_state[js['msg']] = True
self.send_to(json.dumps({
'type': 'login',
'success': True,
'msg': ' , /ls , /help ( / )'
}), cs)
# ,
self.broadcast_system_msg(js['msg'] + " ")
else:
self.send_to(json.dumps({
'type': 'login',
'success': False,
'msg': ' '
}), cs)
# ,
elif js['type'] == "broadcast":
if self.__user_name_to_broadcast_state[self.__socket_to_user_name[cs]]:
self.broadcast(js['msg'], cs)
else:
self.send_to(json.dumps({
'type': 'broadcast',
'msg': ' '
}), cs)
elif js['type'] == "ls":
self.send_to(json.dumps({
'type': 'ls',
'msg': self.get_all_login_user_info()
}), cs)
elif js['type'] == "help":
self.send_to(json.dumps({
'type': 'help',
'msg': self.__help_str
}), cs)
elif js['type'] == "sendto":
self.single_chatting(cs, js['nickname'], js['msg'])
elif js['type'] == "ignore":
self.exchange_ignore_state(cs)
def exchange_ignore_state(self, cs):
if cs in self.__socket_to_user_name:
state = self.__user_name_to_broadcast_state[self.__socket_to_user_name[cs]]
if state:
state = False
else:
state = True
self.__user_name_to_broadcast_state.pop(self.__socket_to_user_name[cs])
self.__user_name_to_broadcast_state[self.__socket_to_user_name[cs]] = state
if self.__user_name_to_broadcast_state[self.__socket_to_user_name[cs]]:
msg = " "
else:
msg = " "
self.send_to(json.dumps({
'type': 'ignore',
'success': True,
'msg': '[TIME : %s]\r
[ SYSTEM ] : %s\r
' % (ctime(), " , " + msg)
}), cs)
else:
self.send_to({
'type': 'ignore',
'success': False,
'msg': ' '
}, cs)
def single_chatting(self, cs, nickname, msg):
if nickname in self.__user_name_to_socket:
msg = '[TIME : %s]\r
[ %s CHATTING TO %s ] : %s\r
' % (
ctime(), self.__socket_to_user_name[cs], nickname, msg)
self.send_to_list(json.dumps({
'type': 'single',
'msg': msg
}), self.__user_name_to_socket[nickname], cs)
else:
self.send_to(json.dumps({
'type': 'single',
'msg': ' '
}), cs)
print(nickname)
def send_to_list(self, msg, *cs):
for i in range(len(cs)):
self.send_to(msg, cs[i])
def get_all_login_user_info(self):
login_list = "[ SYSTEM ] ALIVE USER : \r
"
for key in self.__socket_to_user_name:
login_list += self.__socket_to_user_name[key] + ",\r
"
return login_list
def send_to(self, msg, cs):
if cs not in self.__socket_list:
self.__socket_list.append(cs)
cs.sendall(bytes(msg, 'utf-8'))
def broadcast_system_msg(self, msg):
data = '[TIME : %s]\r
[ SYSTEM ] : %s\r
' % (ctime(), msg)
js = json.dumps({
'type': 'system_msg',
'msg': data
})
#
for i in range(len(self.__socket_list)):
if self.__socket_list[i] in self.__socket_to_user_name:
self.__socket_list[i].sendall(bytes(js, 'utf-8'))
def broadcast(self, msg, cs):
data = '[TIME : %s]\r
[%s] : %s\r
' % (ctime(), self.__socket_to_user_name[cs], msg)
js = json.dumps({
'type': 'broadcast',
'msg': data
})
# ,
for i in range(len(self.__socket_list)):
if self.__socket_list[i] in self.__socket_to_user_name \
and self.__user_name_to_broadcast_state[self.__socket_to_user_name[self.__socket_list[i]]]:
self.__socket_list[i].sendall(bytes(js, 'utf-8'))
def main():
server = PyChattingServer()
server.start_session()
main()
import json
import threading
from socket import *
is_login = False
is_broadcast = True
class ClientReceiveThread(threading.Thread):
__buf = 1024
def __init__(self, cs):
super(ClientReceiveThread, self).__init__()
self.__cs = cs
def run(self):
self.receive_msg()
def receive_msg(self):
while True:
msg = self.__cs.recv(self.__buf).decode('utf-8')
if not msg:
break
js = json.loads(msg)
if js['type'] == "login":
if js['success']:
global is_login
is_login = True
print(js['msg'])
elif js['type'] == "ignore":
if js['success']:
global is_broadcast
if is_broadcast:
is_broadcast = False
else:
is_broadcast = True
print(js['msg'])
else:
if not is_broadcast:
print("[ ]")
print(js['msg'])
class ClientSendMsgThread(threading.Thread):
def __init__(self, cs):
super(ClientSendMsgThread, self).__init__()
self.__cs = cs
def run(self):
self.send_msg()
#
def send_msg(self):
while True:
js = None
msg = input()
if not is_login:
js = json.dumps({
'type': 'login',
'msg': msg
})
elif msg[0] == "@":
data = msg.split(' ')
if not data:
print(" ")
break
nickname = data[0]
nickname = nickname.strip("@")
if len(data) == 1:
data.append(" ")
js = json.dumps({
'type': 'sendto',
'nickname': nickname,
'msg': data[1]
})
elif msg == "/help":
js = json.dumps({
'type': 'help',
'msg': None
})
elif msg == "/ls":
js = json.dumps({
'type': 'ls',
'msg': None
})
elif msg == "/i":
js = json.dumps({
'type': 'ignore',
'msg': None
})
else:
if msg[0] != '/':
js = json.dumps({
'type': 'broadcast',
'msg': msg
})
if js is not None:
self.__cs.sendall(bytes(js, 'utf-8'))
def main():
buf = 1024
# , ,
address = ("127.0.0.1", 12231)
cs = socket(AF_INET, SOCK_STREAM, 0)
cs.connect(address)
data = cs.recv(buf).decode("utf-8")
if data:
print(data)
receive_thread = ClientReceiveThread(cs)
receive_thread.start()
send_thread = ClientSendMsgThread(cs)
send_thread.start()
while True:
pass
main()총결산
이 구현 채 팅 방 에서 저 는 json 형식의 문자열 정 보 를 사용 하여 작성 한 프로 토 콜 입 니 다.아마도 더 간단 한 방식 으로 실현 할 수 있 을 것 입 니 다.
사실 이 채 팅 방 은 가장 기본 적 인 socket 프로 그래 밍 의 실현 방안 이자 네트워크 에 속 하 는 비교적 간단 한 작성 입 니 다.
이상 이 바로 본 고의 모든 내용 입 니 다.여러분 의 학습 에 도움 이 되 고 저 희 를 많이 응원 해 주 셨 으 면 좋 겠 습 니 다.
이 내용에 흥미가 있습니까?
현재 기사가 여러분의 문제를 해결하지 못하는 경우 AI 엔진은 머신러닝 분석(스마트 모델이 방금 만들어져 부정확한 경우가 있을 수 있음)을 통해 가장 유사한 기사를 추천합니다:
로마 숫자를 정수로 또는 그 반대로 변환그 중 하나는 로마 숫자를 정수로 변환하는 함수를 만드는 것이었고 두 번째는 그 반대를 수행하는 함수를 만드는 것이었습니다. 문자만 포함합니다'I', 'V', 'X', 'L', 'C', 'D', 'M' ; 문자열이 ...
텍스트를 자유롭게 공유하거나 복사할 수 있습니다.하지만 이 문서의 URL은 참조 URL로 남겨 두십시오.
CC BY-SA 2.5, CC BY-SA 3.0 및 CC BY-SA 4.0에 따라 라이센스가 부여됩니다.
좋은 웹페이지 즐겨찾기
