使用常規列表，我可以根據對象屬性對列表進行排序：queue.sort(key=lambda weed: (weed.x_coord), reverse=True)但是，對于多處理隊列，這是不可能的，那么我如何用多處理隊列完成相同的排序呢？或者，如果我想最終對隊列進行排序，則最好避免多進程隊列？要求是隊列/列表應該是線程安全和進程安全的，因為隊列/列表將由兩個并行運行的線程填充。將對象插入共享隊列的兩個進程（p1 和 p2）將繼續與從隊列中讀取的第三個進程（狀態機）一起運行（請參閱下面的代碼）。即狀態機進程不會等待 p1 和 p2 進程結束。到目前為止的實現：import multiprocessingclass Weed(): x=None y=None def __init__(self,x,y): self.x=x self.y=ydef p1(q): """ Function that inserts weed in the shared queue """ # append squares of mylist to queue q.put(Weed(10.1,7.3)) q.put(Weed(8.3,2.8)) q.put(Weed(5.1,4.2)) q.put(Weed(15.4,5.0))def p2(q): """ Function that inserts weed in the shared queue """ # append squares of mylist to queue q.put(Weed(25.1,1)) q.put(Weed(1.3,1)) q.put(Weed(9.1,1)) q.put(Weed(13.4,1))def state_machine(q): """ Function that sorts the queue (w.r.t x-coord.) and prints it out """ print("Queue elements:") while not q.empty(): q.sort(key=lambda x: (x.x), reverse=True) # Gives error - print(q.get().x) print("Queue is now empty!")if __name__ == "__main__": # creating multiprocessing Queue q = multiprocessing.Queue() # creating new processes p1 = multiprocessing.Process(target=p1, args=(q,)) p2 = multiprocessing.Process(target=p2, args=(q,)) p3 = multiprocessing.Process(target=state_machine, args=(q,)) # running process p1 to generate some weeds p1.start() # running process p2 to generate some weeds p2.start() # running process p3 to sort the weed queue (by x coord.) and print them out p3.start()
1 回答
子衿沉夜
TA貢獻1828條經驗 獲得超3個贊
在您的示例中，3 個進程不會同時運行（在啟動下一個進程之前啟動它們并加入它們），我將假設實際情況確實具有并發性。
但要小心：在實際情況下，空隊列并不意味著其他任務已完成。您將需要另一種同步機制。
我的建議是回退到函數內的常規列表，并在元素到達時將元素從多處理隊列傳輸到列表。然后，您可以按順序對列表和打印元素進行排序。您不會遇到并發問題，因為內部列表僅由運行 的線程修改。state_machinestate_machine
def state_machine(q):
"""
Function that sorts the queue (w.r.t x-coord.) and prints it out
"""
print("Queue elements:")
internal_queue = []
# here, we assume that all tasks have finished working.
# if it is not the case, you should add a barrier to wait
while not q.empty():
internal_queue.append(q.get())
internal_queue.sort(key=lambda item: (item.x), reverse=True)
for elt in internal_queue:
print(elt.x)
print("Queue is now empty!")
程序打?。?/p>
Queue elements:
25.1
15.4
13.4
10.1
9.1
8.3
5.1
1.3
Queue is now empty!
[編輯]
在實際方案中，您不希望在開始打印之前等待使用者完成。但是，您必須在兩個問題之間找到折衷方案：
如果你在開始消費元素之前等待太長時間，你基本上會回到等待生產者完成。
如果您使用隊列中的元素太快（即在它們到達后立即打印它們），則您的隊列在大部分時間里往往是空的，并且排序不再有太大意義。
那里沒有（恕我直言）最佳解決方案，這里有一個命題，其中內部隊列定期更新，同時讓生產者有時間完成工作：
def state_machine(q):
"""
Function that sorts the queue (w.r.t x-coord.) and prints it out
"""
internal_queue = []
def update_internal_queue():
while not q.empty():
internal_queue.append(q.get())
internal_queue.sort(key=lambda item: (item.x), reverse=True)
# wait a bit for the beginning of the elements to arrive
time.sleep(5)
update_internal_queue()
print("Queue elements:")
while internal_queue:
time.sleep(1) # we can optionally wait a bit before each print
update_internal_queue() # see if other elements arrived in the meantime
print(internal_queue.pop(0).x)
print("Queue is now empty!")
添加回答
舉報
0/150
提交
取消