在Python中通过TCP发送/接收多个消息 [英] Sending/Receiving Multiple Message over TCP in Python
本文介绍了在Python中通过TCP发送/接收多个消息的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！
问题描述
我想在服务器和客户端之间发送/接收消息多个TCP消息。例如：
I want to send/receive messages multiple TCP messages between the server and the client. For Example:
服务器：
ip=""
port=8888
buffersize=1024
s=socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind((myIP, myPort))
s.listen(1)
(cl, adress) = s.accept()
cl.send("hi!".encoded())
msg=cl.recv(bufferSize).decode()
msg=msg[0]
print(" Received message is '{}'".format(msg))
客户：
ip=""
port=8888
s=socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect(ip,port)
msg=s.recv(1024).decode()
print(" {} ".format(msg))
#sending to server
s.send("OK!".encode())
他们可以相互沟通其他。
我想发送第二条消息，并且希望循环接收第二条消息。
They can communicate with each other. I want to send a second message and I want to receive second message in a loop.
推荐答案
我写了一篇证明这一点的代码。撰写内容在这里： https://aknirala.github.io/maintaining/python3TCP.html
代码在这里： https://github.com/aknirala/python3TCP
I wrote a code demonstrating this. Writeup is here: https://aknirala.github.io/maintaining/python3TCP.html Code is here: https://github.com/aknirala/python3TCP
我发现人们（无心的人）删除了我之前发布的链接（即： https://realpython.com/python-sockets/ ）说：如果链接更改，则答案将无效。但是我不明白为什么需要删除，因为在那之后，没有答案！！留下评论来解释问题会很好。无论如何，这里是代码（服务器要执行1000次任务，每次10次。
I found that people (heart-less-ly) deleted the previous link I posted (which was: https://realpython.com/python-sockets/) saying that: if link changes then answer will become invalid. But I don't get why the need to delete, as after that, there was no answer!!. Leaving a comment to explain stuff would have been fine. Anyways here is the code (server wants to do 1000 tasks 10 times each.
服务器端：
#!/usr/bin/env python
# coding: utf-8
# In[ ]:
import socket #For TCP communication.
import types #For types.SimpleNamespace
import selectors #For selectors.DefaultSelector()
import time
import random
# In[ ]:
totalTasks = 1000
repAllTasks = 10
d = {}
for i in range(totalTasks):
d[i] = 0
msgStart = "crrrazy_msg_START"
msgEnd = "crrrazy_msg_END" #Hoping that this will not be in the message.
verbose = False
#Instead of having start and end tokens, a better way is to have headers which will
#have size of the message.
# In[ ]:
sel = selectors.DefaultSelector()
# In[ ]:
def accept_wrapper(sock):
'''
This function will accept the new clients, and will register them.
'''
conn, addr = sock.accept()
print("\n\n accepted connection from", addr)
conn.setblocking(False) #If we want multiple clients we need to do this.
#Below we are assigning three things to each clinet connection, if we want more details
#like, when was the last time clinet communicated we need to pass one more argument here
# and later, ensure that it gets updated.
data = types.SimpleNamespace(addr=addr, inb="", outb=b"")
#Above notice that inb is not binary. This is becasue our code will be simpler
#if we'll keep it string (as a single message will be recieved over multiple
#iterations). But we need to convert it from binary stream before appending it there.
#In server we have made a loop and are senidng data till all is send, so outb can
#be removed, as it is never used.
events = selectors.EVENT_READ | selectors.EVENT_WRITE
sel.register(conn, events, data=data)
# In[ ]:
host, port = "<<<ENTER YOUR MACHINE IP>>>", 65432#sys.argv[1], int(sys.argv[2]) #TODO: Chnage to your machine IP
lsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
lsock.bind((host, port))
lsock.listen()
print("listening on", (host, port))
lsock.setblocking(False)
sel.register(lsock, selectors.EVENT_READ, data=None)
# In[ ]:
underProcess = {}
for i in range(totalTasks):
underProcess[i] = 0
ctr = 0
lastRetCtr = 0
def getNextTask(tD):
'''
tD is a dictionary which has, key values of taskID and units which has been completed.
This function will update the d, accordingly and will then return the next taskID to be done
if tD is None, we are just checking what is the next value to be returned, so no actual
changes are made to any variable.
'''
global d
global underProcess
global repAllTasks
global ctr
global lastRetCtr
#print('Got tD as: ',tD)
#Updating the task dictionary d, and also removing that from underProcess
if tD is not None:
for k in tD:
underProcess[k] -= 1
d[k] += tD[k]
lastRetCtr += 1
ctr = lastRetCtr
for k in d:
if d[k] + underProcess[k] < repAllTasks:
if tD is not None:
underProcess[k] += 1
return k
for k in underProcess:
if underProcess[k] > 0:
#print('k:',k,'underProcess[k]: ',underProcess[k], ' ctr: ',ctr, 'lastRetCtr: ',lastRetCtr)
if tD is not None:
ctr += 1
if ctr - lastRetCtr > 10: #This is just a simple mechanism to resend stuff
# when some unprocess task has not been done.
# If till 10 iterations no client returns then it will
# resend the task (not an ideal mechanism)
if ctr - lastRetCtr > 15:
ctr = lastRetCtr
print('Returning: ',k)
return k
else:
if tD is not None:
time.sleep(1)
return -2
return -1
# In[ ]:
tmpGlobalMsg = "" #These two are defined global coz I don't know how to work with
tmpGlobalTD = None #exec with local variables. :-( Let me know how to get rid of them
ctr = 0
def service_connection(key, mask):
#print('In service_connection')
global d
global msgStart
global msgEnd
global tmpGlobalMsg
global tmpGlobalTD
global ctr
global verbose
sock = key.fileobj
data = key.data
if mask & selectors.EVENT_READ:
recv_data = sock.recv(1024)
if recv_data:
data.inb += recv_data.decode()
eI = data.inb.find(msgEnd) #Is the end str found? If so we have recievd entire message
if eI >= 0:
sI = data.inb.find(msgStart)
if sI < 0 or sI > eI: #Ideally throw an exception here.
sI = 0
else:
sI += len(msgStart)
msg = data.inb[sI:eI]
data.inb = data.inb[eI + len(msgEnd):]
tmpGlobalTD = None
msg = 'tmpGlobalTD = '+msg
exec(msg, locals(), globals())
#Below, we update, how much unit of task has been done by client
nJob = getNextTask(tmpGlobalTD)
print('From : ',sock.getpeername(), 'nJob is: ', nJob)
toSend = msgStart + str(nJob) + msgEnd
toSend = str.encode(toSend) #We need to encode it befre sending
while len(toSend) > 0:
if mask & selectors.EVENT_WRITE:
sent = sock.send(toSend)
toSend = toSend[sent:]
if verbose:
print('Sent: ',sent)
else: #Not needed could be removed.
print('Not ready to send!!!! THIS SHOULD NOT HAPPEN (But who knows!!)')
time.sleep(0.0001)
else: #This will be true when client will close connection. We can even set
print("\n\n closing connection to", data.addr) # a timer here.
sel.unregister(sock)
sock.close()
else:
if verbose:
time.sleep(0.2) #So that output is not flooded
print('Not read ready !!!!', end=" ") #Thre are clients which have registerd
# But no one has sent anything for the server to read.
# In[ ]:
while True:
events = sel.select(timeout=None)
if verbose:
sleepT = 1 - len(events)*0.1
if sleepT <= 0:
sleepT = 0.01
time.sleep(sleepT)
print('NUmber of clinets is: ',len(events))
for key, mask in events:
if key.data is None:
accept_wrapper(key.fileobj)
else:
service_connection(key, mask)
nT = getNextTask(None)
if nT < 0:
if nT == -2:
print('Status: ',underProcess)
time.sleep(1)
continue
print('All done.underProcess ', underProcess)
for key, mask in events:
data = key.data
sock = key.fileobj
print("\n\n closing connection to", data.addr) # a timer here.
sel.unregister(sock)
sock.close()
break
客户端：
#!/usr/bin/env python
# coding: utf-8
# In[ ]:
import socket
import selectors
import types
import time
import random
sel = selectors.DefaultSelector()
msgStart = "crrrazy_msg_START"
msgEnd = "crrrazy_msg_END"
verbose = False
# In[ ]:
def start_connections(host, port):
server_addr = (host, port)
print("\n\nstarting connection to", server_addr)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setblocking(False)
sock.connect_ex(server_addr) #connect_ex() is used instead of connect() since connect() would immediately raise a BlockingIOError exception. connect_ex() initially returns an error indicator, errno.EINPROGRESS, instead of raising an exception while the connection is in progress.
events = selectors.EVENT_READ | selectors.EVENT_WRITE
data = types.SimpleNamespace(
#connid=connid, #In case multiple connections are made from same machine
# we can assign an ID to them.
#Other fields, if other protocols like fixed message lengths etc needs to be
#used then more variables could be added here to keep track of those.
inb="", #Not a binary stream as it will be easier to handle str
)
sel.register(sock, events, data=data)
# In[ ]:
tempTaskID = None
def sendRecv(sel, tS = {}, recv = True, disconnect = False):
'''
This function is written in such a way that one can send a message to server
and server will send a return message and the function will return that.
Since the function may wait endlessly if server does not return it, after some
delay, we are assuming that server didn't listened, so we are resending the message.
This is not an ideal solution, as server might get the message again and falsely
assume that task has been done. This can be fixed by providing unique key to each job,
and IMHO must be done in actual code.
sel: contains the connection to server.
tS: dictionary to send. By default it is empty, which in our case we consider as
blank message.
recv: a flag to indicate, if we are expecting a return message from server.
In our case, this will always be True.
disconnect: When True, this function will close the connection.
'''
global msgStart
global msgEnd
global tempTaskID
global verbose
waitAttempts = 0
while True:
events = sel.select(timeout=1)
if events:
toProcess = True #To ensure that there is only one connection
# If everything goes correctly we don't need this.
# and ideally code involving this shouldnot run.
# Thus all code involving this could be removed,
for key, mask in events: #Idealy we should have just one value in events.
sock = key.fileobj
data = key.data
if not toProcess:
print('Repeated!!!!! ', key.index, ' closing it.')
print('THIS SHOULD NOT HAVE BEEN EXECUTED. But what can we say...')
sel.unregister(sock)
sock.close()
print('Disconnected extra socket.')
continue
toProcess = False
if disconnect: #This time function hsa been called to disconnect the connection.
# Please note, in case we would have been managing multiple
# connections, this would have been added to data when calling
# SimpleNamespace while registering, so that we know which socket
# to close.
sel.unregister(sock)
sock.close()
print('Disconnected')
return {}
msg = msgStart + str(tS) + msgEnd
toSend = str.encode(msg)
if waitAttempts <= 0: #It'll not send again till waitAttempts becomes less
# This is added in case, server does not recieve the message
# for the first time. So after waiting for few times
# ,in our case 10 seconds, as done by sleep in each iteration
# we'll resend the message.
# In practice, this could even happen when message from server
# is lost, thus a better mechanism to detect that should be there
while len(toSend) > 0:
if mask & selectors.EVENT_WRITE:
sent = sock.send(toSend)
if verbose:
print('sent: ',sent, 'From: ',sock.getsockname(),' Cont: ',toSend[:sent])
toSend = toSend[sent:]
else:
print('Not ready to write!!. THIS SHOULD NOT HAVE HAPPENED, but what do I know..')
time.sleep(0.0001)
waitAttempts = 10 #Try again after these many iterations of outer while
else:
waitAttempts -= 1
if not recv: #Return not expected.
return {}
if verbose:
print('wtoR...', end=' ')#For illustration we are SoPing
# wtoR: waiting to READ.
msg = ''
if mask & selectors.EVENT_READ:
recv_data = sock.recv(1024) # Should be ready to read
if recv_data:
data.inb += recv_data.decode()
#print('Now we got:::: ',data.outb,' len: ',len(data.outb), 'and find: ',data.outb.find(msgEnd))
else:
if verbose:
print('NotR..', end=' ')#For illustration we are SoPing
# NotR: Not ready to READ. (Waiting for server to send message)
time.sleep(1) #By giving delay here we will wait for 10 seconds for
# server to reply.
eI = data.inb.find(msgEnd)
if eI >= 0:
sI = data.inb.find(msgStart)
if sI < 0 or sI > eI: #Ideally throw an exception here.
sI = 0
else:
sI += len(msgStart)
msg = data.inb[sI:eI]
if len(msg) <= 0:
print('ERROR, this definitely looks like a protocol failure. Terminating connection, by returning -1')
#We are expecting that server must alway reply an integer eithre
#representing taskID or -1 to terminate.
return -1
data.inb = data.inb[eI + len(msgEnd):] #Simply removing the part
# of the message we have processed. In this case, if everything
# is as planned, this should always set data.inb to empty string.
try:
return int(msg)
except:
print('Integer not returned. Val return is:',msg,'. Protocol break.')
return -1
else:
if verbose:
print('No end on: ', data.inb, end='')
else:
print('No events!!! THIS SHOULD NOT HAPPEN...')
if not sel.get_map():
print('It\'s likely that no socket is opened OR server is not there.')
break
return {}
# In[ ]:
host, port = "<<ENTER YOUR SERVER IP>>", 65432#sys.argv[1:4]
start_connections(host, int(port))
# In[ ]:
d = sendRecv(sel)
while True:
if d <0:
if d == -2:
print('d is: ',d)
time.sleep(0.5)
d = sendRecv(sel)
continue
sendRecv(sel, disconnect=True)
print('Disconnected as server said target achieved',d)
break
tS = {}
tS[d] = random.randint(1,2) #Random number between 1 and 2
print('Sending: ',tS)
d = sendRecv(sel, tS)
# In[ ]:
#sendRecv(sel, disconnect=True)
# In[ ]:
这篇关于在Python中通过TCP发送/接收多个消息的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
查看全文
