python发送byte数据组到tcp的server问题

 更新时间:2023年09月06日 16:06:05   作者:橙木  
这篇文章主要介绍了python发送byte数据组到tcp的server问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教

python发送byte数据组到tcp的server

前一段时间需要和一个tcp服务端进行数据交互,有约定好的数据报文格式,但是是以十六进制形式定义的的,所以在测试数据反馈的时候用python写了一个byte[]数据发送的tcp clinet端的测试demo

代码如下:

from socket import *
import struct
import time
import sys
def init(x):#初始化需要发送的list数组
    lists = [[0]*8 for i in range(7)]#定义list的行列大小
    for i in range(7):
        lists[i].append(0)#填充0
        lists[i][3] = 5
        lists[i][4] = int(x)#填充序号
    lists[0][7] = 16#填充指令id
    lists[1][7] = 17
    lists[2][7] = 18
    lists[3][7] = 19
    lists[4][7] = 20
    lists[5][7] = 24
    lists[6][7] = 25
    return lists
def main():
    # 1.创建tcp_client_socket 套接字对象
    tcp_client_socket = socket(AF_INET,SOCK_STREAM)
    # 作为客户端,主动连接服务器较多,一般不需要绑定端口
    # 2.连接服务器
    tcp_client_socket.connect(("127.0.0.1",7001))
    a = sys.argv[1]
    while True:
        for i in range(7):
            print(init(a)[i])
            """无限循环可以实现无限发送"""
            # 3.向服务器发送数据
            cmd = init(a)[i]
            to_server = ""
            for i in range(len(cmd)):
                to_server += chr(cmd[i])
            print(repr(to_server))
            print("发送的消息为:",to_server.encode())
            tcp_client_socket.send(to_server.encode())# 在linux中默认是utf-8编码
        # 在udp协议中使用的sendto() 因为udp发送的为数据报,包括ip port和数据,           # 所以sendto()中需要传入address,而tcp为面向连接,再发送消息之前就已经连接上了目标主机
        #time.sleep(1)
        # 4.接收服务器返回的消息
            recv_data = tcp_client_socket.recv(1024)  # 此处与udp不同,客户端已经知道消息来自哪台服务器,不需要用recvfrom了
            if recv_data:
                print("返回的消息为:",recv_data)
            else:
                print("对方已离线。。")
                break
    tcp_client_socket.close()
if __name__ == '__main__
	main()

python tcp server-client

基本思路

  • 1、指定IP、端口号;
  • 2、绑定;
  • 3、开启监听;
  • 4、接受连接创建socket;
  • 5、收发数据

tcp_server

# tcp_server
# coding=utf-8
# !/usr/bin/env python
import socket
import time
import threading
serverIP = "0.0.0.0"
serverPort = 10620
clientSocketList = []  # 放每个客户端的socket
def init_server():
    sockServer = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sockServer.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server_addr = (serverIP, serverPort)
    sockServer.bind(server_addr)
    sockServer.listen(10)
    print(server_addr)
    print("The server has started, waiting for the client to connect ......")
    return sockServer
def accept_client(sockServer):
    while True:
        time.sleep(0.1)
        clientSock, addr = sockServer.accept()
        strTime = time.strftime('%Y-%m-%d %H:%M:%S')
        strPrint = 'connected from: {}'.format(addr)
        strPrint = strTime + strPrint
        print(strPrint)
        clientSock.setblocking(0)
        clientSocketList.append([clientSock, addr])
def server_Send(direction):
    while True:
        time.sleep(0.1)
        try:
            sendByte = bytes(direction)
            for clientInfo in clientSocketList:
                currClient, addr = clientInfo
                currClient.sendall(sendByte)
                print("to {}, send <{}> ".format(addr, sendByte))
        except Exception as e:
            clientSocketList.remove(clientInfo)
            continue
def server_Recv():
    while True:
        time.sleep(0.1)
        for clientInfo in clientSocketList:
            # print(client.getsockname())
            # print(client.getpeername())
            currClient, addr = clientInfo
            try:
                dataRecv = currClient.recv(1024)
            except Exception as e:
                continue
            if not dataRecv:
                clientSocketList.remove(clientInfo)
                print("currClient{} has closeed.\n".format(addr))
                continue
            try:
                direction = float(dataRecv)
                strTime = time.strftime('%Y-%m-%d %H:%M:%S')
                strRecv = "from {} recv len={}, data={}".format(addr, len(dataRecv), direction)
                print(strRecv)
            except Exception as e:
                print(e)
                pass
if __name__ == '__main__':
    sockServer = init_server()
    threadCheckClient = threading.Thread(target=accept_client, args=(sockServer, ))  # 子线程
    # threadCheckClient.setDaemon(True)
    threadCheckClient.start()
    threadSend = threading.Thread(target=server_Recv)  # 子线程
    # threadSend.setDaemon(True)
    threadSend.start()

tcp_client

# tcp_client
# coding=utf-8
# !/usr/bin/env python
import socket
import sys, time
from threading import Thread
serverIP = '127.0.0.1'
serverPort = 10620 
def init_client():
  tcp_client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  try:
    tcp_client.connect((serverIP, serverPort))
  except socket.error:
    print('fail to setup socket connection')
  return tcp_client
def client_Send(tcp_client):
    while True:
        time.sleep(0.1)
        time.sleep(1)
        try:
            strTime = time.strftime('%Y-%m-%d %H:%M:%S')
            strTime = str(12345.5678909)
            sendBytes =strTime.encode()
            tcp_client.sendall(sendBytes)
            print(sendBytes)
        except Exception as e:
            break
def client_Recv(tcp_client):
    while True:
        time.sleep(0.1)
        try:
            dataRecv = tcp_client.recv(1024)  # 到这里程序继续向下执行
        except Exception as e:
            continue
        if not dataRecv:
            break
        else:
            strTime = time.strftime('%Y-%m-%d %H:%M:%S')
            strRecv = "from server recv len={}, data={}".format( len(dataRecv), dataRecv)
            strPrint = strTime + strRecv
            print(strPrint)
if __name__ == "__main__":
    tcp_client = init_client()
    threadSend = Thread(target=client_Send, args=(tcp_client, ))  # 子线程
    threadSend.start()

总结

以上为个人经验,希望能给大家一个参考,也希望大家多多支持脚本之家。

相关文章

  • python游戏测试工具自动化遍历游戏中所有关卡

    python游戏测试工具自动化遍历游戏中所有关卡

    这篇文章主要为大家介绍了python游戏测试工具自动化遍历游戏中所有关卡示例详解,有需要的朋友可以借鉴参考下,希望能够有所帮助,祝大家多多进步,早日升职加薪
    2022-06-06
  • Python实现比较两个列表(list)范围

    Python实现比较两个列表(list)范围

    这篇文章主要介绍了Python实现比较两个列表(list)范围,本文根据一道题目实现解决代码,本文分别给出题目和解答源码,需要的朋友可以参考下
    2015-06-06
  • 一文详解Python中三元运算符的用法以及避坑指南

    一文详解Python中三元运算符的用法以及避坑指南

    这篇文章主要为大家详细介绍了Python中的三元运算符(条件表达式),这是一种简洁的条件判断语法结构,文中的示例代码讲解详细,感兴趣的小伙伴可以了解下
    2026-01-01
  • python 数据挖掘算法的过程详解

    python 数据挖掘算法的过程详解

    这篇文章主要介绍了python 数据挖掘算法,首先给大家介绍了数据挖掘的过程,基于sklearn主要的算法模型讲解,给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下
    2022-02-02
  • 使用Python实现ELT统计多个服务器下所有数据表信息

    使用Python实现ELT统计多个服务器下所有数据表信息

    这篇文章主要介绍了使用Python实现ELT统计多个服务器下所有数据表信息,ETL,是英文Extract-Transform-Load的缩写,用来描述将数据从来源端经过抽取(extract)、转换(transform)、加载(load)至目的端的过程,需要的朋友可以参考下
    2023-07-07
  • Python进行MySQL数据备份与增删改查操作实战指南

    Python进行MySQL数据备份与增删改查操作实战指南

    Python是一种强大且易于学习的编程语言,这篇文章主要为大家详细介绍了mysql数据库备份以及利用pymysql模块进行数据库增删改查的相关操作
    2025-07-07
  • Python实现读取文本文件并转换为pdf

    Python实现读取文本文件并转换为pdf

    这篇文章主要为大家详细介绍了如何使用Python简便快捷地完成TXT文件到PDF文档的转换,满足多样化的文档处理需求,感兴趣的小伙伴可以参考下
    2024-04-04
  • Python图像处理之图片拼接和堆叠案例教程

    Python图像处理之图片拼接和堆叠案例教程

    这篇文章主要介绍了Python图像处理之图片拼接和堆叠案例教程,本篇文章通过简要的案例,讲解了该项技术的了解与使用,以下就是详细内容,需要的朋友可以参考下
    2021-07-07
  • Python之Numpy 常用函数总结

    Python之Numpy 常用函数总结

    这篇文章主要介绍了Python之Numpy 常用函数总结,文章围绕主题展开详细的内容介绍,具有一定的参考价值,需要的小伙伴可以参考一下
    2022-07-07
  • 基于Python实现倒计时工具

    基于Python实现倒计时工具

    这篇文章主要为大家详细介绍了基于Python实现倒计时工具,文中示例代码介绍的非常详细,具有一定的参考价值,感兴趣的小伙伴们可以参考一下
    2022-08-08

最新评论