永远不会到达最后一个线程

Question

首先，我将解释我的代码是如何工作的。我有 3 个相互交互的模块：2 个模块通过套接字连接到一个模块并发送 UDP 帧。单个模块接收 UDP 帧，将它们保存到队列中，然后另一个函数将队列作为输入并进行一些处理。

我正在运行在不同终端中发送 UDP 帧的模块。我想运行接收 UDP 帧的函数和在不同线程中对保存的帧进行处理的函数。为此，我使用了线程和队列包。但是我没能同时运行所有线程；它总是卡在第二个线程中，永远不会到达最后一个线程。

这是我的代码：

send_1.py：

import socket

import pickle

import time

def send_frame():

    UDP_IP = "127.0.0.1"

    UDP_PORT = 5005

    MESSAGE = {'x': 0.20, 'y': 0.2, 'z': 0.2}

    MESSAGE = pickle.dumps(MESSAGE)

    print(type(MESSAGE))

    print("UDP target IP:", UDP_IP)

    print("UDP target port:", UDP_PORT)

    print("message:", MESSAGE)

    sock = socket.socket(socket.AF_INET,  # Internet

                         socket.SOCK_DGRAM)  # UDP

    while True:

        sock.sendto(MESSAGE, (UDP_IP, UDP_PORT))

        time.sleep(5)

send_frame()

send_2.py:

import socket

import pickle

import time

def send_frame():

    UDP_IP = "127.0.0.1"

    UDP_PORT = 5006

    # MESSAGE = b"Hello, World!"

    MESSAGE = {'x': 2.20, 'y': 2.2, 'z': 2.2}

    MESSAGE = pickle.dumps(MESSAGE)

    print(type(MESSAGE))

    print("UDP target IP:", UDP_IP)

    print("UDP target port:", UDP_PORT)

    print("message:", MESSAGE)

    sock = socket.socket(socket.AF_INET,  # Internet

                         socket.SOCK_DGRAM)  # UDP

    while True:

        sock.sendto(MESSAGE, (UDP_IP, UDP_PORT))

        time.sleep(5)

send_frame()

这是接收帧的代码，将它们保存到队列然后处理它们。

Answer 1

我设法通过使用asyncio解决了这个问题我对代码进行了一些更改，以便我现在只有方法而没有类。所以代码如下：send_1.pyimport socketimport pickleimport timedef send_frame():    UDP_IP = "127.0.0.1"    UDP_PORT = 5005    MESSAGE = {'x': 0.20, 'y': 0.2, 'z': 0.2}    MESSAGE = pickle.dumps(MESSAGE)    print(type(MESSAGE))    print("UDP target IP:", UDP_IP)    print("UDP target port:", UDP_PORT)    print("message:", MESSAGE)    sock = socket.socket(socket.AF_INET,  # Internet                         socket.SOCK_DGRAM)  # UDP    while True:        sock.sendto(MESSAGE, (UDP_IP, UDP_PORT))        time.sleep(5)send_frame()  send_2.pyimport socketimport pickleimport timedef send_frame():    UDP_IP = "127.0.0.1"    UDP_PORT = 5006    # MESSAGE = b"Hello, World!"    MESSAGE = {'x': 2.20, 'y': 2.2, 'z': 2.2}    MESSAGE = pickle.dumps(MESSAGE)    print(type(MESSAGE))    print("UDP target IP:", UDP_IP)    print("UDP target port:", UDP_PORT)    print("message:", MESSAGE)    sock = socket.socket(socket.AF_INET,  # Internet                         socket.SOCK_DGRAM)  # UDP    while True:        sock.sendto(MESSAGE, (UDP_IP, UDP_PORT))        time.sleep(5)send_frame()现在有了接收帧的代码，将它们保存到队列中然后处理它们。 接收.pyimport asyncioimport queueimport socketimport pickleimport timeasync def receive_frame1(q_1):    UDP_IP = "127.0.0.1"    UDP_PORT = 5005    sock = socket.socket(socket.AF_INET,  # Internet                         socket.SOCK_DGRAM)  # UDP    sock.bind((UDP_IP, UDP_PORT))    while True:        data, addr = sock.recvfrom(1024)  # buffer size is 1024 bytes        data_1 = pickle.loads(data)        print('data_1:', data_1)        ts_1 = time.time()        frame_1 = {'data': data_1, 'timestamp': ts_1}        q_1.put(frame_1)        await asyncio.sleep(0)async def receive_frame2(q_2):    UDP_IP = "127.0.0.1"    UDP_PORT = 5006    sock = socket.socket(socket.AF_INET,  # Internet                         socket.SOCK_DGRAM)  # UDP    sock.bind((UDP_IP, UDP_PORT))    while True:        data, addr = sock.recvfrom(1024)  # buffer size is 1024 bytes        data_2 = pickle.loads(data)        print('data_2:', data_2)        ts_2 = time.time()        frame_2 = {'data': data_2, 'timestamp': ts_2}        q_2.put(frame_2)        await asyncio.sleep(0)async def get_decision(queue_1, queue_2, delta_x, delta_y):    while True:        print('queue_1:', queue_1)        print('queue_2:', queue_2)        frame_1 = queue_1.get()        frame_2 = queue_2.get()        data_1 = frame_1['data']        data_1_ts = frame_1['timestamp']        data_2 = frame_2['data']        data_2_ts = frame_2['timestamp']        decision = 'Unknown'        while time.time() < data_1_ts + 3 and time.time() < data_2_ts + 3:            if (data_2['x'] - delta_x <= data_1['x'] <= data_2['x'] + delta_x and                    data_2['y'] - delta_y <= data_1['y'] <= data_2['y'] + delta_y):                decision = 'Correct'                break            else:                decision = 'Wrong'                break        print('ts:', data_1_ts)        print('ts2:', data_2_ts)        print(decision)        print('#' * 32)        await asyncio.sleep(0)        if __name__ == '__main__':    q_1 = queue.Queue()    q_2 = queue.Queue()    asyncio.ensure_future(receive_frame1(q_1))    asyncio.ensure_future(receive_frame2(q_2))    asyncio.ensure_future(get_decision(q_1, q_2, 3.5, 3.5))    loop = asyncio.get_event_loop()    loop.run_forever()asyncio 帮助我异步运行线程并让它们持续运行loop.run_forever()