使用python socket搭建Client测试平台

概述

1 背景

2 Client功能实现

2.1 何谓Client

2.2 代码功能介绍

2.3 代码实现

2.3.1 代码介绍

2.3.2 代码内容

3 测试

3.1 PC上创建Server

3.2 同一台PC上运行Client

3.2.1 建立连接

3.2.2 测试数据交互

3.3 Linux 环境下运行Client

3.3.1 建立连接

3.3.2 通信功能测试

4 注意问题

概述

本文主要介绍使用Python的Socket搭建一个Server/Client平台，用于测试通信链路的可用性。笔者分别在Windows和linux环境下验证了其可行性，并记录下整个的验证过程。

1 背景

在嵌入式网络功能开发过程中，需要在PC上搭建一个Server，client一般由基于嵌入式功能开发的板卡来实现。在板卡运行功能之前，有必要对PC上的Server进行测试，保证板卡能成功的连接到PC上的Server。因为有的时候，不是Server搭建好了，Cilent就能方便的连上去。可能防火墙等因数，也会引起Client连接Server总是失败。要解决这些问题，前期的测试是非常有必要的，使用Python的Socket接口，能快速的搭建这样的一个平台，且python语言可以运行在不同的平台上。

本文主要介绍使用pyhton实现一个Client,使其运行在不同的平台上，以验证网络链路是否可以正常通信。

2 Client功能实现

2.1 何谓Client

Socket客户端是一个可以向服务器发送请求并接收服务器响应的程序。它使用网络套接字来建立与服务器的连接，并通过套接字进行数据的发送和接收。

Socket客户端的主要功能包括：

建立连接：Socket客户端通过指定服务器的IP地址和端口号来建立与服务器的连接。连接建立成功后，可以开始进行数据的发送和接收。
发送数据：Socket客户端可以向服务器发送数据。它可以将数据封装成一个消息或者数据包，并通过套接字发送给服务器。发送数据的方式可以是一次性发送所有数据，也可以是分块逐步发送。
接收数据：Socket客户端可以从服务器接收数据。它通过套接字接收服务器发送的数据，并进行处理。接收数据的方式可以是一次性接收所有数据，也可以是逐步接收。
处理错误：Socket客户端可以处理与服务器通信过程中出现的错误。这些错误可能包括连接失败、发送数据失败、接收数据失败等。客户端可以通过捕捉异常或者处理错误码来处理这些错误。
断开连接：Socket客户端可以断开与服务器的连接。当不再需要与服务器通信时，可以通过关闭套接字来断开连接。

2.2 代码功能介绍

1）在PC上创建一个Server

2）运行多个Client, 并可以和Server之间能进行发送和接收数据

2.3 代码实现

2.3.1 代码介绍

代码第9行：定义server IP 和端口号

代码第13行：定义接收数据的线程函数

代码第15行：通过socket接收server发送到client的数据

代码第16行：判断数据长度

代码第22行：创建Seocket，选用TCP通信模式

代码第26行：连接Server

代码第28行：创建处理接收数据的线程

代码第29行：启动线程

代码第36行：读取输入数据

代码第37行：client发送数据给server

代码第41行：client断开连接

2.3.2 代码内容

详细代码：

##!/usr/bin/python3
# -*- coding: utf-8 -*-
# @descriptor : client test based on socket
# @Time       : 2024/04/19 20:40
# @Author     : ming fei.tang
import socket
import threading

SERVER_IP = ('192.168.1.5', 19999)
BUFFER_SIZE = 1024


def handler_revMessage(client, address):
    while True:
        data_from_client: str = client.recv(1024).decode("UTF-8")
        if len(data_from_client) > 0:
            print(f"receive the message from server {address} ：{data_from_client}")


def client_socket():
    ip, port = SERVER_IP
    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        print(f'connect to the server: {ip}:{port} ...')

        client.connect((ip, port))
        print(f'connect to {ip}:{port} ready ')
        t1 = threading.Thread(target=handler_revMessage, args=(client, SERVER_IP))
        t1.start()
    except Exception as e:
        print(f'connect to {ip}:{port} fail ')
        client.close()
        assert False, e

    while True:
        answer = input('input the message：')
        client.sendto(answer.encode(), SERVER_IP)
        if 'quit' in answer:
            break

    client.close()
    print(f'disconnect {ip}:{port}  server')


if __name__ == '__main__':
    client_socket()