Python으로 간단한 채팅 프로그램 작성 튜토리얼
구현 아이디어
x01 서버 구축
우선, 서버 측에서 소켓은 메시지를 수락하는 데 사용됩니다. 소켓 요청이 수락될 때마다 새 스레드가 열려 메시지의 배포 및 수락을 관리합니다. 동시에 채팅방의 다양한 기능을 처리하기 위해 모든 스레드를 관리하는 핸들러가 있습니다
x02 클라이언트 설정
클라이언트 설정은 서버보다 훨씬 간단합니다. 클라이언트는 메시지를 보내고 특정 규칙에 따라 특정 문자를 입력하면 됩니다. 따라서 클라이언트 측에서는 두 개의 스레드만 사용하면 됩니다. 하나는 메시지 수신 전용입니다. other는 전송 전용입니다.
우리가 하나를 사용하지 않는 이유는 하나만 사용하면 메시지가 수신될 때 메시지를 받는 사람이 메시지를 보내기 전에 차단된 상태가 되기 때문입니다. 메시지 보내기도 마찬가지이므로 이 두 기능을 한 곳에 구현하면 계속 메시지를 주고받을 방법이 없게 됩니다
구현 방법
서버측 구현
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\n')
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\n'
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\n" \
"输入/ls,即可获得所有登陆用户信息\r\n" \
"输入/h,即可获得帮助\r\n" \
"输入@用户名 (注意用户名后面的空格)+消息,即可发动单聊\r\n" \
"输入/i,即可屏蔽群聊信息\r\n" \
"再次输入/i,即可取消屏蔽\r\n" \
"所有首字符为/的信息都不会发送出去"
__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\n[ SYSTEM ] : %s\r\n' % (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\n[ %s CHATTING TO %s ] : %s\r\n' % (
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\n"
for key in self.__socket_to_user_name:
login_list += self.__socket_to_user_name[key] + ",\r\n"
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\n[ SYSTEM ] : %s\r\n' % (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\n[%s] : %s\r\n' % (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()위 내용은 Python으로 간단한 채팅 프로그램 작성 튜토리얼의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!
핫 AI 도구
Undress AI Tool
무료로 이미지를 벗다
Undresser.AI Undress
사실적인 누드 사진을 만들기 위한 AI 기반 앱
AI Clothes Remover
사진에서 옷을 제거하는 온라인 AI 도구입니다.
Stock Market GPT
더 현명한 결정을 위한 AI 기반 투자 연구
인기 기사
뜨거운 도구
메모장++7.3.1
사용하기 쉬운 무료 코드 편집기
SublimeText3 중국어 버전
중국어 버전, 사용하기 매우 쉽습니다.
스튜디오 13.0.1 보내기
강력한 PHP 통합 개발 환경
드림위버 CS6
시각적 웹 개발 도구
SublimeText3 Mac 버전
신 수준의 코드 편집 소프트웨어(SublimeText3)
Python에서 요구 사항에서 패키지를 설치하는 방법
Sep 18, 2025 am 04:24 AM
종속성 패키지를 설치하려면 pipinstall-rrequirements.txt를 실행하십시오. 충돌을 피하고, 파일 경로가 올바른지 확인하고 PIP가 업데이트되었는지 확인하고 필요한 경우 설치 동작을 조정하기 위해-no-deps 또는 --user와 같은 옵션을 사용하는 것이 좋습니다.
Pytest로 Python 코드를 테스트하는 방법
Sep 20, 2025 am 12:35 AM
Python은 Python의 간단하고 강력한 테스트 도구입니다. 설치 후 테스트 파일은 이름 지정 규칙에 따라 자동으로 발견됩니다. 어설 션 테스트를 위해 test_로 시작하여 기능을 작성하고 @pytest.fixture를 사용하여 재사용 가능한 테스트 데이터를 생성하고 pytest.raises를 통해 예외를 확인하고 지정된 테스트 실행 및 여러 명령 줄 옵션을 지원하며 테스트 효율성을 향상시킵니다.
파이썬에서 명령 줄 인수를 처리하는 방법
Sep 21, 2025 am 03:49 AM
theargparsemoduleisecomedendedway handlecommand-lineargumentsinpython, robustparsing, typevalidation, helpmessages, anderrorhandling; audys.argvforsimplecaseSrequiringMinimalSetup을 제공합니다.
BIP 란 무엇입니까? 비트 코인의 미래에 왜 그렇게 중요한가?
Sep 24, 2025 pm 01:51 PM
목차 비트 코인 개선 제안 (BIP)이란 무엇입니까? BIP가 왜 그렇게 중요한가? 역사적 BIP 프로세스는 비트 코인 개선 제안 (BIP)에 어떻게 효과가 있습니까? BIP 유형 신호는 무엇이며 광부는 어떻게 보내나요? BITCOIN 개선 제안 또는 "BIP"라는 시스템을 통해 2011 년부터 BIP 결론에 대한 빠른 시험의 Taproot 및 Cons는 2011 년부터 개선되었습니다. 비트 코인 개선 제안 (BIP)은 비트 코인이 일반적으로 어떻게 발전 할 수 있는지에 대한 지침을 제공하며, 두 가지 유형의 BIP가 있습니다. 그 중 두 가지 유형은 비트 코인의 기술 변화와 관련이 있습니다.
초보자부터 전문가까지 : 10 개의 필수 무료 공개 데이터 세트 웹 사이트
Sep 15, 2025 pm 03:51 PM
데이터 과학의 초보자에게는 "경험"에서 "업계 전문가"로의 도약의 핵심은 지속적인 실습입니다. 실습의 기초는 풍부하고 다양한 데이터 세트입니다. 다행히 인터넷에는 무료 공개 데이터 세트를 제공하는 많은 웹 사이트가 있으며 기술을 향상시키고 기술을 연마하는 데 유용한 리소스입니다.
Python에서 @ContextManager 데코레이터를 사용하여 컨텍스트 관리자를 어떻게 만들 수 있습니까?
Sep 20, 2025 am 04:50 AM
import@contextManagerFromContextLibandDefineAgeneratorFunctionThatYieldSActlyOnce, whereCodeBeforeYieldActSasEnterAndErandCodeftertyield (바람직하게는) ACTSAS__EXIT __
빅 데이터 분석에 적합한 컴퓨터를 선택하는 방법은 무엇입니까? 고성능 컴퓨팅을위한 구성 안내서
Sep 15, 2025 pm 01:54 PM
빅 데이터 분석은 멀티 코어 CPU, 대용량 메모리 및 계층 스토리지에 중점을 두어야합니다. 코어 수와 단일 코어 성능을 고려하여 Amdepyc 또는 Ryzenthreadripper와 같은 멀티 코어 프로세서가 선호됩니다. 메모리는 64GB로 시작하는 것이 좋습니다. 데이터 무결성을 보장하기 위해 ECC 메모리가 권장됩니다. 스토리지는 NVMESSD (시스템 및 핫 데이터), SATASSD (공통 데이터) 및 HDD (콜드 데이터)를 사용하여 전반적인 처리 효율을 향상시킵니다.
파이썬에서 일상적인 작업을위한 자동화 스크립트 작성 방법
Sep 21, 2025 am 04:45 AM
repetitivetasksworthautomating, suchasorganizingfileSordingEmails, succurnificernificanttime.2.useappropriatepythonlibraries -libraries -libraries likeos, shutil, glob, smtplib, beautifulsoup, andseleniumforfileoperations, w
