个人中心
gpt4 book ai didi

python实现简单的聊天小程序

转载 作者：qq735679552 更新时间：2022-09-27 22:32:09 36 4
gpt4 key购买 nike

CFSDN坚持开源创造价值，我们致力于搭建一个资源共享平台，让每一个IT人在这里找到属于你的精彩世界.

这篇CFSDN的博客文章python实现简单的聊天小程序由作者收集整理，如果你对这篇文章有兴趣，记得点赞哟.

概要

这是一个使用python实现一个简单的聊天室的功能,里面包含群聊,私聊两种聊天方式.实现的方式是使用套接字编程的一个使用tcp协议 c/s结构的聊天室 。

实现思路

x01 服务端的建立 。

首先,在服务端,使用socket进行消息的接受,每接受一个socket的请求,就开启一个新的线程来管理消息的分发与接受,同时,又存在一个handler来管理所有的线程,从而实现对聊天室的各种功能的处理 。

x02 客户端的建立 。

客户端的建立就要比服务端简单多了,客户端的作用只是对消息的发送以及接受,以及按照特定的规则去输入特定的字符从而实现不同的功能的使用,因此,在客户端这里,只需要去使用两个线程,一个是专门用于接受消息,一个是专门用于发送消息的 。

至于为什么不用一个呢,那是因为,只用一个的话,当接受了消息,在发送之前接受消息的处于阻塞状态,同理,发送消息也是,那么要是将这两个功能放在一个地方实现,就会导致没有办法连续发送或者接受消息了 。

实现方式

服务端实现 。

python实现简单的聊天小程序

python实现简单的聊天小程序

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
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()

客户端的实现 。

python实现简单的聊天小程序

python实现简单的聊天小程序

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
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编程的实现方案,也是一些属于网络方面的比较简单的编写吧 。

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持我.

原文链接：https://blog.csdn.net/a591243801/article/details/80916355 。

最后此篇关于python实现简单的聊天小程序的文章就讲到这里了,如果你想了解更多关于python实现简单的聊天小程序的内容请搜索CFSDN的文章或继续浏览相关文章，希望大家以后支持我的博客！ 。

36 4 0
文章推荐： jQuery实现朋友圈查看图片
文章推荐： Python3中最常用的5种线程锁实例总结
文章推荐： python aiohttp的使用详解
文章推荐： python获取对象信息的实例详解
qq735679552
个人简介

我是一名优秀的程序员,十分优秀！

滴滴打车优惠券免费领取
滴滴打车优惠券
全站热门文章
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com