UDP Service
import socket
import threading
import time
class Accept(threading.Thread):
def __init__(self, ip, port):
threading.Thread.__init__(self)
ipaddress = (ip, port)
client = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
client.bind(ipaddress)
self.client = client
def run(self):
while True:
time.sleep(1)
try:
data, adder = self.client.recvfrom(4096)
if data:
string = data.decode('utf-8')
print('服务端 ', adder, string)
self.client.sendto('1'.encode('utf-8'), adder)
if 'Q' == string or 'q' == string:
self.client.close()
break
else:
break
except BaseException:
print('错误')
accept = Accept('127.0.0.1', 5000)
accept.start()
UDP Client
import socket
BUFF_SIZE = 1024
client = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
ipaddress = ('127.0.0.1', 5000)
lop = True
while lop:
text = input('>>').strip().encode('utf-8')
client.sendto(text, ipaddress)
data, adder = client.recvfrom(BUFF_SIZE)
print('客户端', data, adder)
if text == 'q' or text == 'Q':
client.close()
lop = False
UDP Test
C:\ProgramData\Anaconda3\python.exe E:/SourceCode/PyCharm/Test/udp/server.py
服务端 ('127.0.0.1', 53580) 123123123
C:\ProgramData\Anaconda3\python.exe E:/SourceCode/PyCharm/Test/udp/client.py
>>123123123
客户端 b'1' ('127.0.0.1', 5000)
>>
|