[我叫以赏]Python制作交互式的服务器与客户端互相通讯(引用SOCKET模块)

 

前言

 

欢迎来到我的教程啊,我是以赏,这么说吧,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 # 声明全局变量a
while 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()

 

效果预览:

 

 

当用户输入错误时,会提示。如下:

 

 

说明:

 

  1. 开启服务器——启动服务器
  2. 查看连接列表——查看用户连接数
  3. 发送数据——发送数据给用户
  4. 日志——查看用户发来的消息、服务器状态等
  5. 停止服务器——关闭服务器
  6. 退出——退出程序

 

按照以上思路,我们先导入模块,将功能一做好,要求:用户可以自由设置绑定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 

发表评论 (0)

后再参与讨论
公告
公告