当前位置：首页 > news >正文

python tcp server client示例代码

news 2025/8/16 12:47:05

功能：实现基本的tcp server端、client端，并引入threading, 保证两端任意链接、断链接，保证两端的稳定运行

IP说明： server不输入IP，默认为本机的IP，client需要输入要链接的server端的IP

端口说明：server, client端保持一致

ADB调试说明：在连接数据线的情况下，PC安装ADB调试工具，android端打开ADB调试权限，输入adb forward tcp:12345 tcp:8888, 其中12345为PC端口， 8888为android端口。参数链接：adb forward 参考

server code:

from socket import *
import threading
import time
from typing import Optionalclass DataSend(threading.Thread):tcp_server = Optional[socket]def __init__(self, port):threading.Thread.__init__(self, name="data send")self.port = portdef run(self):# Creating the socket objectself.tcp_server = socket(AF_INET, SOCK_STREAM)# Binding to socketself.tcp_server.bind(('', self.port))  # Host will be replaced/substitued with IP, if changed and not running on host# Starting TCP listenerprint('begin listen')self.tcp_server.listen(3)print('end listen')while True:# Starting the connectionprint('begin accept')clientsocket, address = self.tcp_server.accept()print(address, 'accepted!')try:while True:msg_bits = clientsocket.recv(1024)print(msg_bits.decode('utf-8'))# Message sent to client after successful connectionmessage = 'hello! Thank you for connecting to the server' + "\r\n"clientsocket.send( message.encode('utf-8'))except ConnectionResetError:continueclientsocket.close()if __name__ == '__main__':data_send = DataSend(12345)data_send.start()while data_send.is_alive():time.sleep(1)

client code:

import threading
from socket import *
import time
from typing import Optionalclass DataRec(threading.Thread):tcp_client: Optional[socket]def __init__(self, ip, port):threading.Thread.__init__(self, name="data rec")self.ip = ipself.port = portself.tcp_client = Nonedef wait_tcp_connect(self):while self.tcp_client is None:time.sleep(1)self.tcp_client = socket(AF_INET, SOCK_STREAM)try:print('try to init client {}:{}'.format(self.ip, self.port))self.tcp_client.connect((self.ip, self.port))print('client inited!')except Exception as e:self.tcp_client = Noneprint("client init failed, waiting for server!")def run(self):self.wait_tcp_connect()msg_buffer = ''while True:try:time.sleep(1)self.tcp_client.send('hello from client'.encode('utf-8'))msg_bits = self.tcp_client.recv(1024*8)if not msg_bits:continuemsg_str = msg_bits.decode('utf-8')print("rec: {}".format(msg_str))except error as msg:print('client rec msg catch error({} - {})'.format(error, msg))self.tcp_client.close()self.tcp_client = Noneself.wait_tcp_connect()msg_buffer = ''# except Exception as e:#     print("client cat other error({})".format(e))if __name__ == '__main__':data_rec = DataRec('127.0.0.1', 12345)data_rec.start()while data_rec.is_alive():time.sleep(1)

查看全文

http://www.lryc.cn/news/159645.html