Wait! Pikachu is collecting the page......
前言
欢迎来到我的教程啊，我是以赏，这么说吧，Python我也在学习并未达到“精通”的地步，一部分呢是自学，一部分是老师“传授”的。但我认为学习Python应该“学以致用”（学其它也一样）。易语言的领域呆久了，出来混混Python（嘻嘻嘻）。感谢知乎平台，就因为这样我才有机会发布这篇文章。以及网上的自学资源少之又少，对新手不友好，而且大部分资源标着“转载”的标识，那么我学会了Python服务器与客户端的通讯，要把它写成文章分享出来!
所用到的材料与材料的讲解
Python:是一种计算机编程语言。是一种面向对象的动态类型语言，最初被设计用于编写自动化脚本(shell)，随着版本的不断更新和语言新功能的添加，越来越多被用于独立的、大型项目的开发。（摘自360百科）
Python中的socket模块:提供了服务器与客户端通讯的条件，它提供了标准的 BSD Sockets API，Python自带。(Socket又称"套接字"，应用程序通常通过"套接字"向网络发出请求或者应答网络请求，使主机间或者一台计算机上的进程间可以通讯)
Python中的threading模块:用于多线程的操作，“threading”是“穿线（线程）”的意思。
另外还需要一款你上手的Python IDE。
思路
网上大部分思路近似与这张图
这种思路只可以，允许一个客户端连接，而且当一个客户端连接后，整个程序会以堵塞的状态继续运行（等待客户时，下面的代码不会运行，下面详细讲解）。
所以我们加上多线程，如下
这个思路图采取了多线程的方法（即支持并行），当线程①开启等待客户连接，连接后又开启一个新的线程，继续等待客户了连接，线程①在与客户交互时，并不妨碍线程②等待连接客户，此类方法有效的避免了堵塞的问题。(小提示：并行，类似于你再玩游戏，有人敲门，你一边玩游戏一边开门接待客人)
线程的简单说明
线程：是操作系统能够进行运算调度的最小单位。它被包含在进程之中，是进程中的实际运作单位。一条线程指的是进程中一个单一顺序的控制流，一个进程中可以并发多个线程，每条线程并行执行不同的任务。进程与线程之间是包含关系。下面是一个比较通俗的图：
当进程启动时（你的程序被运行时），主线程就开始了，我们来看一看下面的图
主线程是当一个进程被创建，于此同时这个进程一个线程也被创建，通常我们叫这个线程叫主线程，主线程像学校中的校长一样，有权限处理学校大事与规定放学时间。主线程还可以开启若干个子线程，校长可以雇几个班主任来上课。当主线程结束时，程序也就结束了（在没有开子线程的条件下）。
子线程通常是辅助主线程的线程。当开启子线程时，主线程仍然可以继续运行（子线程与主线程并行）。
Python中线程的实现
请看下方代码逐句解析：
# -*- coding: utf-8 -*import threading # 引入 threading 模块 在下文作用于多线程import time # 映入 time 模块 在下文作用于延迟a = 0 # 声明一个变量 在下文用作于技术def calculate():"""这个函数用户测试线程"""global a # 声明全局变量awhile True: # 指定一个循环a = a + 1 # a为记录循环的次数if __name__ == "__main__": # 代码开始cal = threading.Thread(target=calculate) # 启动线程的写法之一cal.start() # 开启线程time.sleep(5) # 延迟5秒print(a) # 查看循环次数结果（5秒后打印）：
（结果因配置而异）
while循环会“暂停”程序，循环不结束，循环的下方代码不执行，相当于把循环所在的线程进入了一个暂停的状态。
cal = threading.Thread(target=calculate) #启动线程的写法之一
cal.start()# 开启线程
重点是这两段代码，对于 Thread 类的注释官方是这样的：
- A class that represents a thread of control.
This class can be safely subclassed in a limited fashion. There are two ways
to specify the activity: by passing a callable object to the constructor, or
by overriding the run() method in a subclass. - 表示控制线程的类。
- 可以通过有限的方式安全地将此类归为一类。有两种方法
- 指定活动：通过将可调用对象传递给构造函数，
- 或通过覆盖子类中的run（）方法。
官方对于“threading.Thread()”的注释:
-
This constructor should always be called with keyword arguments. Arguments are:*group* should be None; reserved for future extension when a ThreadGroupclass is implemented.*target* is the callable object to be invoked by the run()method. Defaults to None, meaning nothing is called.*name* is the thread name. By default, a unique name is constructed ofthe form "Thread-N" where N is a small decimal number.*args* is the argument tuple for the target invocation. Defaults to ().*kwargs* is a dictionary of keyword arguments for the targetinvocation. Defaults to {}.If a subclass overrides the constructor, it must make sure to invokethe base class constructor (Thread.__init__()) before doing anythingelse to the thread.
-
始终应使用关键字参数来调用此构造函数。参数为：*group* 应该为 None; 当线程组类已实现。*target* 是run()可调用的对象，默认为“无”，表示不调用任何内容。*name* 是线程名。默认情况下，唯一的名称由“Thread-N”的形式，其中N是一个小的十进制数。*args* 是目标函数的参数——元组。默认为（）。*kwargs* 是目标函数的参数——字典。默认为{}。如果子类重写构造函数，则必须确保调用基类构造函数（Thread）
上述中target是目标函数的名称，name为线程标记名，args为传入数据的元组,kwargs为传入的字典。
使用 threading.Thread(target=calculate) 呢，要使用start或run启动线程，这里就不多加赘述，线程最后再看一下传入参数吧。
# -*- coding: utf-8 -*
import threading # 引入 threading 模块 在下文作用于多线程
import time # 映入 time 模块 在下文作用于延迟
a = 0 # 声明一个变量 在下文用作于技术
def calculate(*args, **kwargs):
"""
这个函数用户测试线程
"""
print(args, kwargs) # 打印参数
if __name__ == "__main__": # 代码开始
cal = threading.Thread(target=calculate, args=("元组", "参数"), kwargs={"字典参数": "参数"}) # 启动线程的写法之一
cal.start() # 开启线程
time.sleep(5) # 延迟5秒
print(a) # 查看循环次数
Python中服务器与客户端通讯的简单例子
这个标题里，我们展现简单的例子（不支持并行），这个标题内会讲的详细一点。
先看第一段代码：
# -*- coding: utf-8 -* import socket # 引入 socket 模块 Server=socket.socket() # 创建一个socket对象
当然，你也或许会看到有些人的代码是这样写的：
socket.socket(socket.AF_INET, socket.SOCK_STREAM)
没错，socket.socket有4个参数，family(int), type(int), proto(int), fileno(bool)
官方在下方给的注释：
-
For user code address family and type values are IntEnum members, butfor the underlying _socket.socket they're just integers. Theconstructor of _socket.socket converts the given argument to aninteger automatically.
-
对于用户代码，地址族和类型值是IntEnum成员，但是对于底层的_socket.socket来说，它们只是整数。的_socket.socket的构造函数将给定参数转换为自动整数。
官方的话对参数的解释起不到大作用，但有两个参数我是可以解决的。
family:地址族，地址族如下
- socket.AF_UNIX ：只能够用于单一的Unix系统进程间通信
- socket.AF_INET ：服务器之间的网络通信（ipv4协议的TCP和UDP）ipv4,默认为这个
- socket.AF_INET6 ：服务器之间的网络通信ipv6
type:套接字类型：
- socket.SOCK_STREAM ：面向连接的TCP，默认为这个
- socket.SOCK_DGRAM ：面向非连接的UDP
解释一下：
TCP你可以理解成用电话号码打电话，你打一个电话给别人，别人接受了，你才可以与他进行沟通交流。
UDP你可以理解成送邮件，你发给别人，别人不用接受，已经收到他的邮箱里了。（说不定时是垃圾邮件哦^o^狗头保命）
寻找资料后：
proto:通讯协议
- IPPROTO_IP 0 IP协议
IPPROTO_ICMP 1 控制消息协议
IPPROTO_IGMP 2 互联网群组管理协议
IPPROTO_GGP 3 网关^ 2（已弃用）
IPPROTO_TCP 6 tcp
IPPROTO_PUP 12 pup
IPPROTO_UDP 17 user datagram protocol
IPPROTO_IDP 22 xns idp
IPPROTO_ND 77 非官方网盘原型
IPPROTO_RAW 255 原始IP数据包
IPPROTO_MAX 256
fileno：文件描述符（不太清楚）
创建套接字完成后，就是要开始绑定与监听端口了：
Server.bind(("127.0.0.1", 23333)) # 绑定IP与端口，IP与端口以元组的数据传播（IP，端口）
Server.listen() # 开始监听
bind用于绑定端口，listen用于开始监听。
然后就是等待客户连接了。
User = Server.accept() # 等待客户(当用户连接时，这段代码执行完毕，返回一个元组，介绍在下方)
request, client_address = User # 数据分析(把客户信息与客户地址分开分别放入request与client_address)
data = request.recv(1024) # 等待接收数据,最大允许长度为1024B
print("客户 ",client_address," 发来消息：",data.decode("GBK")) # 解码输出
User=Server.accept()返回一个元组，元组构成如下(客户信息,客户地址)，而用于操作的是客户信息部分，客户地址组成：(IP,端口)
data = request.recv(1024)等待客户数据的发送与接收，最大允许长度为1024字节。
我们使用NetAssist（网络调试助手）来测试这段代码，返回如下：
当然，发送的数据不止一条，修改一下：
while True:
data = request.recv(1024) # 等待接收数据
if not data:
break
print("客户 ",client_address," 发来消息：",data.decode("GBK")) # 解码输出
增添一个whlie循环，结束的条件是发过来的数据为无（没有数据）。
完整代码：
# -*- coding: utf-8 -*
import socket # 引入 socket 模块
Server = socket.socket() # 创建一个socket对象
Server.bind(("127.0.0.1", 23333)) # 绑定IP与端口，IP与端口以元组的数据传播（IP，端口）
Server.listen() # 开始监听
User = Server.accept() # 等待客户
request, client_address = User # 数据分析
while True:
data = request.recv(1024) # 等待接收数据
if not data:
break
print("客户 ",client_address," 发来消息：",data.decode("GBK")) # 解码输出
# request.send(data) #发送数据
忘记说了，发送数据方式：
request.send(data) 用户数据.send(数据) # 注意数据要编码
现在我们来制作简单客户端，
有了服务器的经验，客户端就简单了。
# -*- coding: utf-8 -* import socket # 引入 socket 模块 Client = socket.socket() # 创建一个socket对象
同样是创建一个socket对象，但是这次不用绑定端口与监听了。
Client.connect(("127.0.0.1", 23333)) # 连接
直接连接服务器。
接收是类似的，不过直接跳过握手这一步，直接接收数据。
data = Client.recv(1024)
print("服务器消息:", data.decode("GBK"))
完整代码：
# -*- coding: utf-8 -*
import socket # 引入 socket 模块
Client = socket.socket() # 绑定
Client.connect(("127.0.0.1", 23333)) # 连接
data = Client.recv(1024)
print("服务器消息:", data.decode("GBK"))
# Client.send(data) # 发送数据
上述主要讲的接收数据的方式，发送数据比较简单，已注释。
开始正式制作
什么？看了这么多才正式开始？不要急，学习上面的铺垫，下面做起来就很简单了。
我们要做的是一个交互式的软件，就要把交互式的结构先做好，如下：
# -*- coding: utf-8 -*
import os
def main(msg: str = None):
if not msg:
msg = "无消息"
os.system("cls") # 清除输出（注意只可以在控制台使用，不要在IDE中调试，可以删除）
print("消息：", msg)
user = input("""WELCOME！
欢迎使用，Python的socket库中的TCP通讯，本页面模仿的是服务端！
功能列表：
------------------------------------
| 1.开启服务器 | 2.查看连接列表 |
------------------------------------
| 3.发送数据 | 4.日志 |
------------------------------------
| 5.停止服务器 |
------------------------------------
| 输入6或exit可以退出程序 |
------------------------------------
Please Enter:""")
if user == "1":
pass
elif user == "2":
pass
elif user == "3":
pass
elif user == "4":
pass
elif user == "5":
pass
elif user == "6" or user == "exit":
pass
else:
pass
main("输入错误！请重新输入！")
return
if __name__ == "__main__":
main()
效果预览：
当用户输入错误时，会提示。如下：
说明：
- 开启服务器——启动服务器
- 查看连接列表——查看用户连接数
- 发送数据——发送数据给用户
- 日志——查看用户发来的消息、服务器状态等
- 停止服务器——关闭服务器
- 退出——退出程序
按照以上思路，我们先导入模块，将功能一做好，要求：用户可以自由设置绑定IP与端口。
代码如下：
# -*- coding: utf-8 -*
import os # 引入 os 模块 (作用为 清空控制台)
import socket # 引入 socket 模块 (作用为 TCP通讯)
import datetime # 引入 datetime 库 （用于 获取时间 ）
Host = "127.0.0.1" # 声明IP变量
Post = 65001 # 声明Post变量
Server = socket.socket() # 创建socket对象
user = input("""WELCOME！
欢迎使用，Python的socket库中的TCP通讯，本页面模仿的是服务端！
功能列表：
------------------------------------
| 1.开启服务器 | 2.查看连接列表 |
------------------------------------
| 3.发送数据 | 4.日志 |
------------------------------------
| 5.停止服务器 |
------------------------------------
| 输入6或exit可以退出程序 |
------------------------------------
Please Enter:""")
if user == "1": # 启动服务器
Host = input("请输入服务器的运行的IP（默认为127.0.0.1），不输入采用默认，"
"（注意：输入的正确性取决于服务器是否成功开启），Please Enter：") # 询问地址
if len(Host) == 0: # 默认
Host = "127.0.0.1"
print("将地址设置为 %s 成功！" % (Host))
Post = input("请输入服务器运行的端口，端口由整数组成（默认为65001），"
"输入错误或端口被占用会导致服务器无法开启！Please Enter：") # 询问端口
if len(Post) == 0: # 默认
Post = "65001"
while Post.isdigit() == False: # 无限询问
print("端口必须是整数，不可以使用英文字母、符号、中文等代替，请重新输入！")
Post = input("请输入服务器运行的端口，端口由整数组成（默认为65001），"
"输入错误或端口被占用会导致服务器无法开启！Please Enter：")
if len(Post) == 0:
Post = "65001"
print("将地址设置为 %s 成功！" % (Post))
Post = int(Post)
print("正在使用已有数据命服务器启动......")
# 检测是否出错
try:
Server = socket.socket() # 引用
Server.bind((Host, Post))
Server.listen()
except (OSError) as error:
information.append("启动服务器失败，错误原因：" + str(error))
main("出现错误~，错误原因： %s 请确认端口是否被占用！" % (error))
return
# information, ip = Server.accept()
# print(information, ip)
#accept_run = threading.Thread(target=accept)
#accept_run.start()
#thread_ident.append(accept_run.ident)
information.append("启动服务器成功！服务器运行在 %s : %d 上" % (Host, Post))
main("已在 %s : %d 上运行服务器，等待客户连接中......" % (Host, Post))
return
前面的用户输入都是铺垫，一定要让用户输入一个正确的端口。
后面就是启动服务器了，启动服务器使用了try语句（你可以理解成尝试启动服务器），如果服务器启动失败，端口占用等，不至于直接软件崩溃。
try语句基本用法为：
try:
命令
except 错误类型 as 变量:
# 变量是错误原因
说明，有一个变量是information(信息)，用于存储日志内容。
接着我们完成启动服务器的接待部分——并行，建立一个用户列表变量
def accept():
global Server
try:
user_information = Server.accept()
except:
return
data = user_information[0]
data2 = user_information[1]
User_list.append(user_information)
ID = len(User_list) - 1
information.append("[%s]用户 %s 上线了！" % (datetime.datetime.now(), data2))
accept_run = threading.Thread(target=accept) # 再运行一个
accept_run.start()
while True:
try:
msg = data.recv(1024)
except:
User_list.remove(user_information)
information.append("[%s]用户 %s 下线了！" % (datetime.datetime.now(), data2))
break
if not msg:
User_list.remove(user_information)
information.append("[%s]用户 %s 下线了！" % (datetime.datetime.now(), data2))
return
else:
try:
information.append("[%s]用户 %s 发来信息：%s" % (datetime.datetime.now(), User_list[ID][1], msg.decode("GBK")))
except:
information.append("[%s]用户 %s 发来的消息无法解码！" % (datetime.datetime.now(), data2))
入口点这样写：
accept_run = threading.Thread(target=accept) accept_run.start()
我们在线程中接待客户一定要记得使用try语句，否则当用户下线、关闭服务器时会报错！
接着我们完善查看连接列表与日志：
elif user=="2":
a = 0
print("-" * 100)
print("当前连接用户如下(总计数量%d):" % (len(User_list)))
for infor in User_list:
a = a + 1
print("用户 %d/%d : %s" % (a, len(User_list), infor[1]))
print("-" * 100)
input("Please Enter...")
日志：
elif user == "4":
print("-" * 100)
print("目前日志个数(总计数量%d):" % (len(information)))
for x in range(len(information)):
print("(%d/%d)" % (x + 1, len(information)), information[x])
print("-" * 100)
input("Please Enter...")
上面的操作就是把用户列表与日志列表有序的打印出来。
完善发送数据：
我采用的是建立一个函数
def send(Int, msg: str):
if Int <= 0:
main("发送失败，原因：没有这个用户ID")
information.append("[%s]发送内容 %s 时失败！" % (datetime.datetime.now(), msg))
return
if len(User_list) < Int:
main("发送失败，原因：没有这个用户ID")
information.append("[%s]发送内容 %s 时失败！" % (datetime.datetime.now(), msg))
return
# if str=="":
# print("没有输入任何东西！")
# return
try:
Bytes = msg.encode("GBK")
User_list[Int - 1][0].send(Bytes)
except BaseException as error:
print("发送失败，原因：%s" % (str(error)))
information.append("[%s]发送内容 %s 给 %s 时失败！" % (datetime.datetime.now(), msg, User_list[Int - 1][1]))
return
main("消息发送成功！")
information.append("[%s]发送内容 %s 给 %s 时成功！" % (datetime.datetime.now(), msg, User_list[Int - 1][1]))
函数中的参数Int指用户列表中用户ID，msg为str类型：消息
入口点：
elif user == "3":
user_ = input("请输入连接用户中的ID：")
if user_.isdigit() == False:
print("错误！用户ID必须是正整数！")
main()
return
elif int(user_) <= 0:
print("用户ID必须是真整数，可以到连接列表查看！")
main()
return
user2 = input("请输入发送内容：")
send(int(user_), user2)
return
完善停止服务器：
elif user == "5":
stop_id = input("确定停止服务器？(1停止 2取消)：")
while stop_id != "1" and stop_id != "2":
stop_id = input("确定停止服务器？(1停止 2取消)：")
if stop_id == "1":
break
elif stop_id == "2":
main()
return
if stop_id == "1":
try:
Server.close()
except BaseException:
main("出现错误无法停止！你确定服务器打开了吗？")
information.append("[%s]服务器停止失败！" % (datetime.datetime.now()))
main()
return
main("停止服务器成功！")
return
退出的话先停止再退出：
try:
Server.shutdown(2)
Server.close()
except BaseException:
return
sys.exit(1)
完整代码：
# -*- coding: utf-8 -*
import os # 引入 os 模块 (作用为 清空控制台)
import sys # 用于关闭
import socket # 引入 socket 模块 (作用为 TCP通讯)
import threading # 引入 threading 库 （作用为 多线程)
import datetime # 引入 datetime 库 （用于 获取时间 ）
Host = "127.0.0.1" # 声明IP变量
Post = 65001 # 声明Post变量
Server = socket.socket() # 创建socket对象
information = []
User_list = []
def accept():
global Host
global Post
global User_list
global information
global Server
try:
user_information = Server.accept()
except:
return
data = user_information[0]
data2 = user_information[1]
User_list.append(user_information)
ID = len(User_list) - 1
information.append("[%s]用户 %s 上线了！" % (datetime.datetime.now(), data2))
accept_run = threading.Thread(target=accept) # 再运行一个
accept_run.start()
while True:
try:
msg = data.recv(1024)
except:
User_list.remove(user_information)
information.append("[%s]用户 %s 下线了！" % (datetime.datetime.now(), data2))
break
if not msg:
User_list.remove(user_information)
information.append("[%s]用户 %s 下线了！" % (datetime.datetime.now(), data2))
return
else:
try:
information.append("[%s]用户 %s 发来信息：%s" % (datetime.datetime.now(), User_list[ID][1], msg.decode("GBK")))
except:
information.append("[%s]用户 %s 发来的消息无法解码！" % (datetime.datetime.now(), data2))
def send(Int, msg: str):
global Host
global Post
global User_list
global information
global Server
if Int <= 0:
main("发送失败，原因：没有这个用户ID")
information.append("[%s]发送内容 %s 时失败！" % (datetime.datetime.now(), msg))
return
if len(User_list) < Int:
main("发送失败，原因：没有这个用户ID")
information.append("[%s]发送内容 %s 时失败！" % (datetime.datetime.now(), msg))
return
# if str=="":
# print("没有输入任何东西！")
# return
try:
Bytes = msg.encode("GBK")
User_list[Int - 1][0].send(Bytes)
except BaseException as error:
print("发送失败，原因：%s" % (str(error)))
information.append("[%s]发送内容 %s 给 %s 时失败！" % (datetime.datetime.now(), msg, User_list[Int - 1][1]))
return
main("消息发送成功！")
information.append("[%s]发送内容 %s 给 %s 时成功！" % (datetime.datetime.now(), msg, User_list[Int - 1][1]))
def main(msg: str = None):
global Host
global Post
global User_list
global information
global Server
"""
交互函数
:param msg:str
"""
if not msg: # 判断消息有无
msg = "无消息"
os.system("cls") # 清除输出（注意只可以在控制台使用，不要在IDE中调试，可以删除）
print("消息：", msg) # 提示消息
user = input("""WELCOME！
欢迎使用，Python的socket库中的TCP通讯，本页面模仿的是服务端！
功能列表：
------------------------------------
| 1.开启服务器 | 2.查看连接列表 |
------------------------------------
| 3.发送数据 | 4.日志 |
------------------------------------
| 5.停止服务器 |
------------------------------------
| 输入6或exit可以退出程序 |
------------------------------------
Please Enter:""")
if user == "1": # 启动服务器
Host = input("请输入服务器的运行的IP（默认为127.0.0.1），不输入采用默认，"
"（注意：输入的正确性取决于服务器是否成功开启），Please Enter：") # 询问地址
if len(Host) == 0: # 默认
Host = "127.0.0.1"
print("将地址设置为 %s 成功！" % (Host))
Post = input("请输入服务器运行的端口，端口由整数组成（默认为65001），"
"输入错误或端口被占用会导致服务器无法开启！Please Enter：") # 询问端口
if len(Post) == 0: # 默认
Post = "65001"
while Post.isdigit() == False: # 无限询问
print("端口必须是整数，不可以使用英文字母、符号、中文等代替，请重新输入！")
Post = input("请输入服务器运行的端口，端口由整数组成（默认为65001），"
"输入错误或端口被占用会导致服务器无法开启！Please Enter：")
if len(Post) == 0:
Post = "65001"
print("将地址设置为 %s 成功！" % (Post))
Post = int(Post)
print("正在使用已有数据命服务器启动......")
# 检测是否出错
try:
Server = socket.socket() # 引用
Server.bind((Host, Post))
Server.listen()
except (OSError) as error:
information.append("启动服务器失败，错误原因：" + str(error))
main("出现错误~，错误原因： %s 请确认端口是否被占用！" % (error))
return
# information, ip = Server.accept()
# print(information, ip)
accept_run = threading.Thread(target=accept)
accept_run.start()
information.append("启动服务器成功！服务器运行在 %s : %d 上" % (Host, Post))
main("已在 %s : %d 上运行服务器，等待客户连接中......" % (Host, Post))
return
elif user == "2":
a = 0
print("-" * 100)
print("当前连接用户如下(总计数量%d):" % (len(User_list)))
for infor in User_list:
a = a + 1
print("用户 %d/%d : %s" % (a, len(User_list), infor[1]))
print("-" * 100)
input("Please Enter...")
main()
return
elif user == "3":
user_ = input("请输入连接用户中的ID：")
if user_.isdigit() == False:
print("错误！用户ID必须是正整数！")
main()
return
elif int(user_) <= 0:
print("用户ID必须是真整数，可以到连接列表查看！")
main()
return
user2 = input("请输入发送内容：")
send(int(user_), user2)
return
elif user == "4":
print("-" * 100)
print("目前日志个数(总计数量%d):" % (len(information)))
for x in range(len(information)):
print("(%d/%d)" % (x + 1, len(information)), information[x])
print("-" * 100)
input("Please Enter...")
main()
return
elif user == "5":
stop_id = input("确定停止服务器？(1停止 2取消)：")
while stop_id != "1" and stop_id != "2":
stop_id = input("确定停止服务器？(1停止 2取消)：")
if stop_id == "1":
break
elif stop_id == "2":
main()
return
if stop_id == "1":
try:
Server.close()
except BaseException:
main("出现错误无法停止！你确定服务器打开了吗？")
information.append("[%s]服务器停止失败！" % (datetime.datetime.now()))
main()
return
main("停止服务器成功！")
return
elif user == "6" or user == "exit":
try:
Server.shutdown(2)
Server.close()
except BaseException:
return
sys.exit(1)
else:
pass
main("输入错误！请重新输入！")
return
if __name__ == "__main__":
main()
客户端代码与服务器代码类似这里就不讲了
import socket # 引入socket （用于 连接服务器 ）
import threading
global Host
global Post
global Con
global Client
information = []
Host = "127.0.0.1"
Post = 65001
Con = False
Client = socket.socket() # 绑定
def acc():
try:
byte = Client.recv(1024)
except:
return
information.append("服务器发来消息：%s" % (byte.decode('GBK')))
run = threading.Thread(target=acc)
run.start()
def main():
global Host
global Post
global Con
global Client
user = input("""欢迎使用socket连接测试软件（客户端）
-------------
1.连接服务器
2.连接状态
3.发送消息
4.程序日志
5.断开连接
6.退出程序
-------------
请输入编号："""
)
if user == "1":
Client = socket.socket() # 绑定
user = input("请输入你要连接的IP地址(未输入，默认采用127.0.0.1)：")
if user == "":
Host = "127.0.0.1"
else:
Host = user
print("成功设置 Host 为 %s " % (Host))
user = input("输入目标主机的端口(默认采用65001)：")
if user == "":
Post = 65001
else:
try:
Post = int(user)
except BaseException:
print("端口无法转化为整数型，默认采用65001！")
Post = 65001
print("成功设置 Post 为 %d " % (Post))
print("开始连接服务器......")
try:
Client.connect((Host, Post))
except BaseException as error:
print("连接过程出现了问题 %s " % (str(error)))
if str(error).find("10061") != -1:
print("遇到了[WinError 10061]问题，可能是因为目标计算机未启动服务器。")
elif str(error).find("11001") != -1:
print("遇到了[WinError 11001]问题，找不到主机。")
information.append("连接服务器时出错！原因： %s " % (str(error)))
Con = False
main()
return
print("连接服务器成功！")
run = threading.Thread(target=acc)
run.start()
information.append("成功连接服务器！数据： %s:%d" % (Host, Post))
Con = True
main()
return
elif user == "2":
print("当前连接状态为 %s . 欲连接地址与端口：%s:%d " % (boolTostr(Con), Host, Post))
elif user == "3":
user = input("请输入你想要发送的内容：")
try:
sth=user.encode("GBK")
Client.send(sth)
except BaseException as error:
print("发送错误！原因：%s" % (str(error)))
information.append("尝试发送数据 %s 时出错！原因： %s" % (user, str(error)))
else:
print("发送成功！")
information.append("成功发送数据 %s " % (user))
main()
return
elif user == "4":
print("-" * 100)
print("目前日志个数(总计数量%d):" % (len(information)))
for x in range(len(information)):
print("(%d/%d)" % (x + 1, len(information)), information[x])
print("-" * 100)
main()
return
elif user == "5":
while True:
user = input("确定断开连接？(1.是的 2.不是)：")
if user == "1":
try:
Client.close()
except:
print("断开失败！")
else:
print("断开成功！")
main()
return
elif user == "2":
return
else:
continue
main()
return
elif user == "6":
try:
Client.close()
except:
raise BaseException("断开失败！强制断开！")
exit()
def boolTostr(bool):
if bool == True:
return "已连接"
else:
return "未连接"
if __name__ == "__main__":
main()
最后，在知乎、BiliBili、以赏的秘密小屋、CSDN上才是我发布的，转载注明链接！
链接: https://pan.baidu.com/s/18ckFDVAO3ybS5GzSQnoXSA 提取码: aaaa