首先，我將解釋我的代碼是如何工作的。我有 3 個相互交互的模塊：2 個模塊通過套接字連接到一個模塊并發(fā)送 UDP 幀。單個模塊接收 UDP 幀，將它們保存到隊列中，然后另一個函數(shù)將隊列作為輸入并進行一些處理。我正在運行在不同終端中發(fā)送 UDP 幀的模塊。我想運行接收 UDP 幀的函數(shù)和在不同線程中對保存的幀進行處理的函數(shù)。為此，我使用了線程和隊列包。但是我沒能同時運行所有線程；它總是卡在第二個線程中，永遠不會到達最后一個線程。這是我的代碼：send_1.py：import 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.py:import 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()這是接收幀的代碼，將它們保存到隊列然后處理它們。
1 回答
胡子哥哥
TA貢獻1825條經(jīng)驗 獲得超6個贊
我設法通過使用asyncio解決了這個問題
我對代碼進行了一些更改，以便我現(xiàn)在只有方法而沒有類。
所以代碼如下：
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()
現(xiàn)在有了接收幀的代碼，將它們保存到隊列中然后處理它們。 接收.py
import asyncio
import queue
import socket
import pickle
import time
async 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 幫助我異步運行線程并讓它們持續(xù)運行l(wèi)oop.run_forever()
添加回答
舉報
0/150
提交
取消