用Python編寫簡單的聊天程式教程
python
小程式
實現想法
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\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中文網其他相關文章！
本網站聲明
本文內容由網友自願投稿，版權歸原作者所有。本站不承擔相應的法律責任。如發現涉嫌抄襲或侵權的內容，請聯絡admin@php.cn
熱AI工具
Undresser.AI Undress
人工智慧驅動的應用程序，用於創建逼真的裸體照片
AI Clothes Remover
用於從照片中去除衣服的線上人工智慧工具。
Undress AI Tool
免費脫衣圖片
Clothoff.io
AI脫衣器
AI Hentai Generator
免費產生 AI 無盡。
熱門文章
R.E.P.O.能量晶體解釋及其做什麼（黃色晶體）
4 週前
By 尊渡假赌尊渡假赌尊渡假赌
R.E.P.O.最佳圖形設置
4 週前
By 尊渡假赌尊渡假赌尊渡假赌
刺客信條陰影：貝殼謎語解決方案
2 週前
By DDD
R.E.P.O.如果您聽不到任何人，如何修復音頻
4 週前
By 尊渡假赌尊渡假赌尊渡假赌
WWE 2K25：如何解鎖Myrise中的所有內容
1 個月前
By 尊渡假赌尊渡假赌尊渡假赌
熱工具
記事本++7.3.1
好用且免費的程式碼編輯器
SublimeText3漢化版
中文版，非常好用
禪工作室 13.0.1
強大的PHP整合開發環境
Dreamweaver CS6
視覺化網頁開發工具
SublimeText3 Mac版
神級程式碼編輯軟體(SublimeText3)
熱門話題
HadiDB：Python 中的輕量級、可水平擴展的數據庫
Apr 08, 2025 pm 06:12 PM
HadiDB：輕量級、高水平可擴展的Python數據庫HadiDB(hadidb)是一個用Python編寫的輕量級數據庫，具備高度水平的可擴展性。安裝HadiDB使用pip安裝：pipinstallhadidb用戶管理創建用戶：createuser()方法創建一個新用戶。 authentication()方法驗證用戶身份。 fromhadidb.operationimportuseruser_obj=user("admin","admin")user_obj.
Python：探索其主要應用程序
Apr 10, 2025 am 09:41 AM
Python在web開發、數據科學、機器學習、自動化和腳本編寫等領域有廣泛應用。 1)在web開發中，Django和Flask框架簡化了開發過程。 2)數據科學和機器學習領域，NumPy、Pandas、Scikit-learn和TensorFlow庫提供了強大支持。 3)自動化和腳本編寫方面，Python適用於自動化測試和系統管理等任務。
2小時的Python計劃：一種現實的方法
Apr 11, 2025 am 12:04 AM
2小時內可以學會Python的基本編程概念和技能。 1.學習變量和數據類型，2.掌握控制流（條件語句和循環），3.理解函數的定義和使用，4.通過簡單示例和代碼片段快速上手Python編程。
Navicat查看MongoDB數據庫密碼的方法
Apr 08, 2025 pm 09:39 PM
直接通過 Navicat 查看 MongoDB 密碼是不可能的，因為它以哈希值形式存儲。取回丟失密碼的方法：1. 重置密碼；2. 檢查配置文件（可能包含哈希值）；3. 檢查代碼（可能硬編碼密碼）。
如何針對高負載應用程序優化 MySQL 性能？
Apr 08, 2025 pm 06:03 PM
MySQL數據庫性能優化指南在資源密集型應用中，MySQL數據庫扮演著至關重要的角色，負責管理海量事務。然而，隨著應用規模的擴大，數據庫性能瓶頸往往成為製約因素。本文將探討一系列行之有效的MySQL性能優化策略，確保您的應用在高負載下依然保持高效響應。我們將結合實際案例，深入講解索引、查詢優化、數據庫設計以及緩存等關鍵技術。 1.數據庫架構設計優化合理的數據庫架構是MySQL性能優化的基石。以下是一些核心原則：選擇合適的數據類型選擇最小的、符合需求的數據類型，既能節省存儲空間，又能提升數據處理速度
如何將 AWS Glue 爬網程序與 Amazon Athena 結合使用
Apr 09, 2025 pm 03:09 PM
作為數據專業人員，您需要處理來自各種來源的大量數據。這可能會給數據管理和分析帶來挑戰。幸運的是，兩項 AWS 服務可以提供幫助：AWS Glue 和 Amazon Athena。
redis怎麼啟動服務器
Apr 10, 2025 pm 08:12 PM
啟動 Redis 服務器的步驟包括：根據操作系統安裝 Redis。通過 redis-server（Linux/macOS）或 redis-server.exe（Windows）啟動 Redis 服務。使用 redis-cli ping（Linux/macOS）或 redis-cli.exe ping（Windows）命令檢查服務狀態。使用 Redis 客戶端，如 redis-cli、Python 或 Node.js，訪問服務器。
redis怎麼讀取隊列
Apr 10, 2025 pm 10:12 PM
要從 Redis 讀取隊列，需要獲取隊列名稱、使用 LPOP 命令讀取元素，並處理空隊列。具體步驟如下：獲取隊列名稱：以 "queue:" 前綴命名，如 "queue:my-queue"。使用 LPOP 命令：從隊列頭部彈出元素並返回其值，如 LPOP queue:my-queue。處理空隊列：如果隊列為空，LPOP 返回 nil，可先檢查隊列是否存在再讀取元素。
