草庐IT

python - 在 Python 中通过 TCP 发送/接收多条消息

coder 2023-09-19 原文

我想在服务器和客户端之间发送/接收多个 TCP 消息。例如:

服务器:

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())

他们可以互相交流。 我想发送第二条消息,我想循环接收第二条消息。

最佳答案

我写了一段代码来证明这一点。文章在这里:https://aknirala.github.io/maintaining/python3TCP.html 代码在这里:https://github.com/aknirala/python3TCP

我发现人们(无情地)删除了我之前发布的链接(即:https://realpython.com/python-sockets/)说:如果链接更改,则答案将失效。但是我不明白为什么需要删除,因为在那之后,没有答案!!留下评论来解释事情就好了。无论如何,这里是代码(服务器想要执行 1000 个任务,每个任务 10 次。

服务器端:

#!/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 - 在 Python 中通过 TCP 发送/接收多条消息,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/51590277/

有关python - 在 Python 中通过 TCP 发送/接收多条消息的更多相关文章

  1. python - 如何使用 Ruby 或 Python 创建一系列高音调和低音调的蜂鸣声? - 2

    关闭。这个问题是opinion-based.它目前不接受答案。想要改进这个问题?更新问题,以便editingthispost可以用事实和引用来回答它.关闭4年前。Improvethisquestion我想在固定时间创建一系列低音和高音调的哔哔声。例如:在150毫秒时发出高音调的蜂鸣声在151毫秒时发出低音调的蜂鸣声200毫秒时发出低音调的蜂鸣声250毫秒的高音调蜂鸣声有没有办法在Ruby或Python中做到这一点?我真的不在乎输出编码是什么(.wav、.mp3、.ogg等等),但我确实想创建一个输出文件。

  2. ruby-on-rails - 如何在 Rails View 上显示错误消息? - 2

    我是rails的新手,想在form字段上应用验证。myviewsnew.html.erb.....模拟.rbclassSimulation{:in=>1..25,:message=>'Therowmustbebetween1and25'}end模拟Controller.rbclassSimulationsController我想检查模型类中row字段的整数范围,如果不在范围内则返回错误信息。我可以检查上面代码的范围,但无法返回错误消息提前致谢 最佳答案 关键是您使用的是模型表单,一种显示ActiveRecord模型实例属性的表单。c

  3. ruby-on-rails - RSpec:避免使用允许接收的任何实例 - 2

    我正在处理旧代码的一部分。beforedoallow_any_instance_of(SportRateManager).toreceive(:create).and_return(true)endRubocop错误如下:Avoidstubbingusing'allow_any_instance_of'我读到了RuboCop::RSpec:AnyInstance我试着像下面那样改变它。由此beforedoallow_any_instance_of(SportRateManager).toreceive(:create).and_return(true)end对此:let(:sport_

  4. jquery - 我的 jquery AJAX POST 请求无需发送 Authenticity Token (Rails) - 2

    rails中是否有任何规定允许站点的所有AJAXPOST请求在没有authenticity_token的情况下通过?我有一个调用Controller方法的JqueryPOSTajax调用,但我没有在其中放置任何真实性代码,但调用成功。我的ApplicationController确实有'request_forgery_protection'并且我已经改变了config.action_controller.consider_all_requests_local在我的environments/development.rb中为false我还搜索了我的代码以确保我没有重载ajaxSend来发送

  5. Python 相当于 Perl/Ruby ||= - 2

    这个问题在这里已经有了答案:关闭10年前。PossibleDuplicate:Pythonconditionalassignmentoperator对于这样一个简单的问题表示歉意,但是谷歌搜索||=并不是很有帮助;)Python中是否有与Ruby和Perl中的||=语句等效的语句?例如:foo="hey"foo||="what"#assignfooifit'sundefined#fooisstill"hey"bar||="yeah"#baris"yeah"另外,类似这样的东西的通用术语是什么?条件分配是我的第一个猜测,但Wikipediapage跟我想的不太一样。

  6. ruby-on-rails - 如何使用 Rack 接收 JSON 对象 - 2

    我有一个非常简单的RubyRack服务器,例如:app=Proc.newdo|env|req=Rack::Request.new(env).paramspreq.inspect[200,{'Content-Type'=>'text/plain'},['Somebody']]endRack::Handler::Thin.run(app,:Port=>4001,:threaded=>true)每当我使用JSON对象向服务器发送POSTHTTP请求时:{"session":{"accountId":String,"callId":String,"from":Object,"headers":

  7. java - 什么相当于 ruby​​ 的 rack 或 python 的 Java wsgi? - 2

    什么是ruby​​的rack或python的Java的wsgi?还有一个路由库。 最佳答案 来自Python标准PEP333:Bycontrast,althoughJavahasjustasmanywebapplicationframeworksavailable,Java's"servlet"APImakesitpossibleforapplicationswrittenwithanyJavawebapplicationframeworktoruninanywebserverthatsupportstheservletAPI.ht

  8. ruby - 使用 Ruby 通过 Outlook 发送消息的最简单方法是什么? - 2

    我的工作要求我为某些测试自动生成电子邮件。我一直在四处寻找,但未能找到可以快速实现的合理解决方案。它需要在outlook而不是其他邮件服务器中,因为我们有一些奇怪的身份验证规则,我们需要保存草稿而不是仅仅发送邮件的选项。显然win32ole可以做到这一点,但我找不到任何相当简单的例子。 最佳答案 假设存储了Outlook凭据并且您设置为自动登录到Outlook,WIN32OLE可以很好地完成此操作:require'win32ole'outlook=WIN32OLE.new('Outlook.Application')message=

  9. Ruby - 如何将消息长度表示为 2 个二进制字节 - 2

    我正在使用Ruby,我正在与一个网络端点通信,该端点在发送消息本身之前需要格式化“header”。header中的第一个字段必须是消息长度,它被定义为网络字节顺序中的2二进制字节消息长度。比如我的消息长度是1024。如何将1024表示为二进制双字节? 最佳答案 Ruby(以及Perl和Python等)中字节整理的标准工具是pack和unpack。ruby的packisinArray.您的长度应该是两个字节长,并且按网络字节顺序排列,这听起来像是n格式说明符的工作:n|Integer|16-bitunsigned,network(bi

  10. 华为OD机试用Python实现 -【明明的随机数】 2023Q1A - 2

    华为OD机试题本篇题目:明明的随机数题目输入描述输出描述:示例1输入输出说明代码编写思路最近更新的博客华为od2023|什么是华为od,od薪资待遇,od机试题清单华为OD机试真题大全,用Python解华为机试题|机试宝典【华为OD机试】全流程解析+经验分享,题型分享,防作弊指南华为o

随机推荐