Python实现TCP监听端口的完整代码示例
本文提供了一段使用Python编写的TCP监听端完整代码。通过socket模块建立连接,利用struct处理数据头,实现了持续接收数据并解析线程ID与错误码的功能。代码包含异常处理机制,支持通过命令行参数指定监听端口,适用于需要稳定监听服务端口的Python开发场景。
本文分享了一个使用Python语言编写TCP监听端的具体实现方案,供开发者参考。
import socket
import time
import sys
import string
import struct
import errno
import binascii
#Definition
ser_ip = 'localhost'
ser_port = 15001
HEADER_LISTENER = "IIII"
split_time = 4
class TcpClient:
def run_srv(self):
sock = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
print ("Trying to connect server...")
addr = (ser_ip, ser_port)
print ("Connecting " + ser_ip + ":" + str(ser_port))
#Connect server
try:
sock.connect(addr)
except Exception,e:
print ("Error:%s" % (e))
sock.close()
sys.exit()
hl = struct.pack(HEADER_LISTENER,0,0,0,0)
header_len = len(hl)
while True:
try:
buf_recv = sock.recv(header_len)
buf_header = buf_recv[0:header_len]
thread_id = struct.unpack("!4I" , buf_header)[0]
err_num = struct.unpack("!4I" , buf_header)[1]
com_num = struct.unpack("!4I" , buf_header)[2]
wait_num = struct.unpack("!4I" , buf_header)[3]
#print("header len %d, recv len %d,buf_header:%s,buf_recv:%s")%(header_len,len(buf_recv),binascii.hexlify(buf_header),binascii.hexlify(buf_recv))
print ("split time:%d")%(split_time)
print ("thread id :%d")%(thread_id)
print ("error nums:%d")%(err_num)
print ("compl nums:%d")%(com_num)
print ("wait nums:%d")%(wait_num)
print ("----------------------")
except Exception,e:
print ("Error:%s" % (e))
sock.close()
sys.exit()
if __name__ == '__main__':
if (len(sys.argv) >= 2):
ser_port = int(sys.argv[1])
client = TcpClient()
client.run_srv()
以上内容为本文的全部介绍,希望能为读者的学习提供实际帮助,同时也欢迎继续支持本站获取更多技术资源。
本站声明
本文内容仅供参考,不代表本站观点。转载或引用请注明出处。如发现内容问题,请通过联系我们页面反馈。













