使用常規(guī)列表，我可以根據(jù)對象屬性對列表進(jìn)行排序：queue.sort(key=lambda weed: (weed.x_coord), reverse=True)但是，對于多處理隊(duì)列，這是不可能的，那么我如何用多處理隊(duì)列完成相同的排序呢？或者，如果我想最終對隊(duì)列進(jìn)行排序，則最好避免多進(jìn)程隊(duì)列？要求是隊(duì)列/列表應(yīng)該是線程安全和進(jìn)程安全的，因?yàn)殛?duì)列/列表將由兩個(gè)并行運(yùn)行的線程填充。將對象插入共享隊(duì)列的兩個(gè)進(jìn)程（p1 和 p2）將繼續(xù)與從隊(duì)列中讀取的第三個(gè)進(jìn)程（狀態(tài)機(jī)）一起運(yùn)行（請參閱下面的代碼）。即狀態(tài)機(jī)進(jìn)程不會等待 p1 和 p2 進(jìn)程結(jié)束。到目前為止的實(shí)現(xiàn)：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貢獻(xiàn)1828條經(jīng)驗(yàn) 獲得超3個(gè)贊
在您的示例中，3 個(gè)進(jìn)程不會同時(shí)運(yùn)行（在啟動下一個(gè)進(jìn)程之前啟動它們并加入它們），我將假設(shè)實(shí)際情況確實(shí)具有并發(fā)性。
但要小心：在實(shí)際情況下，空隊(duì)列并不意味著其他任務(wù)已完成。您將需要另一種同步機(jī)制。
我的建議是回退到函數(shù)內(nèi)的常規(guī)列表，并在元素到達(dá)時(shí)將元素從多處理隊(duì)列傳輸?shù)搅斜怼Ｈ缓螅梢园错樞驅(qū)α斜砗痛蛴≡剡M(jìn)行排序。您不會遇到并發(fā)問題，因?yàn)閮?nèi)部列表僅由運(yùn)行 的線程修改。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!")
程序打印：
Queue elements:
25.1
15.4
13.4
10.1
9.1
8.3
5.1
1.3
Queue is now empty!
[編輯]
在實(shí)際方案中，您不希望在開始打印之前等待使用者完成。但是，您必須在兩個(gè)問題之間找到折衷方案：
如果你在開始消費(fèi)元素之前等待太長時(shí)間，你基本上會回到等待生產(chǎn)者完成。
如果您使用隊(duì)列中的元素太快（即在它們到達(dá)后立即打印它們），則您的隊(duì)列在大部分時(shí)間里往往是空的，并且排序不再有太大意義。
那里沒有（恕我直言）最佳解決方案，這里有一個(gè)命題，其中內(nèi)部隊(duì)列定期更新，同時(shí)讓生產(chǎn)者有時(shí)間完成工作：
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!")
添加回答
舉報(bào)
0/150
提交
取消