ข้อความ Multiplexing RabbitMQ

Question

ตัวอย่างเช่นฉันมีแหล่งข้อมูล 4 แหล่งที่เผยแพร่เครื่องมือวัด ฉันต้องการมัลติเพล็กซ์ / รวมข้อความเหล่านี้ทั้งหมดในคิว / การแลกเปลี่ยนเดียว

--------+----+----+----+----+        -------+---------+----+---------+---------+
Source1 | M1 | M2 | M3 |    |   =>   Result | M1 | M4 | M2 | M3 | M6 | M5 | M7 |
Source2 | M4 |    |    | M5 |
Source3 |    |    | M6 |    |
Source4 |    |    |    | M7 |

For each queue:
 * Read one message
 * Publish message to the Result queue

มีวิธี "ดั้งเดิม" ในการทำเช่นนี้ใน RabbitMQ หรือฉันควรเขียน Consumer / Publisher ของตัวเอง?

แก้ไข 1

ตัวอย่างบางส่วนเพื่อชี้แจงสมมติว่าหลังจากเวลาที่ฉันมี

     Processing "window"
        +-+
Source1 |X|XXXXXXXXXXXXX
Source2 |Y|YYYYYYY
Source3 |Z|ZZZZZZZZZZ
Source4 |W|WW
        +-+

และหลังจากนั้น

     Processing "window"
           +-+
Source1 XXX|X|XXXXXXXXXX
Source2 YYY|Y|YYYY
Source3 ZZZ|Z|ZZZZZZZ
Source4 WWW| |
           +-+

และหลังจากนั้น

     Processing "window"
                 +-+
Source1 XXXXXXXXX|X|XXXX
Source2 YYYYYYYY | |
Source3 ZZZZZZZZZ|Z|Z
Source4 WWW      | |
                 +-+

ผลการสั่งซื้อจะเป็น: X Y Z W X Y Z W X Y Z W X Y Z X Y Z X Y Z X Y Z X Y Z X Z X Z X Z X X X

X, Y, Z, W จากนั้น X, Y, Z, W แล้ว X, Y, Z, W แล้ว X, Y, Z ... X, Z ...

ด้วยวิธีนี้แม้ว่าแหล่งที่มาจะ "สแปม" ข้อความอื่น ๆ ทั้งหมดจากแหล่งอื่นก็มีโอกาสที่จะถูกใช้ไป

ด้วยเหตุผลทางเทคนิค / การเงินฉันต้องใช้เพียง 1 ข้อความต่อครั้ง

ผู้บริโภคช้ากว่าผู้ผลิต แต่ผู้ผลิตเผยแพร่มาก แต่เป็นครั้งคราว

หากแต่ละแหล่งเผยแพร่ไปยังการแลกเปลี่ยนที่ผูกกับคิวเดียวกันผลลัพธ์อาจเป็นXXXXXXXXXXXXXX YYYYYYYY ZZZZZZZZZZZ WWWหรือXXXXX Y XXXXX YYY XXX YYYY ZZZZZZZZZZZ WWW(ขึ้นอยู่กับอัตราการเผยแพร่ของแต่ละแหล่ง)

Answer 1

ฉันคิดว่าสิ่งที่คุณต้องการสามารถทำได้เพียงแค่เรียกใช้สคริปต์เดียวที่สมัครรับคิวทั้งหมด

ข้อกำหนดหลักคือการใช้เธรดแอปพลิเคชันเดียวเพื่อจัดการข้อความทั้งหมดไม่ว่าจะมาจากคิวใดก็ตาม สิ่งที่ดูเหมือนจะแตกต่างกันไปขึ้นอยู่กับภาษาและไลบรารีไคลเอ็นต์ที่คุณใช้ - หากคุณใช้ PHP คุณจะต้องหลีกเลี่ยงไม่ให้เป็นเธรดเดียว แต่อาจมีไลบรารีไคลเอ็นต์อยู่บ้าง ที่ถือว่าการโทรกลับแต่ละครั้งอยู่ในเธรดของผู้ปฏิบัติงานที่แยกจากกันและคุณจะต้องมีทรัพยากรที่ใช้ร่วมกันเพื่อบล็อก

ในแง่ของสิ่งต่างๆของ RabbitMQ คุณจะต้อง:

• ลงทะเบียนสมัครสมาชิกเซิร์ฟเวอร์เพื่อส่งข้อความถึงคุณด้วยbasic.consume; โดยทั่วไปแล้วจะแนะนำมากกว่าการสำรวจความคิดเห็นอย่างชัดเจนด้วยbasic.getอย่างไรก็ตาม
• ใช้ "ช่อง" เดียวสำหรับการbasic.consumeโทรทั้งหมด
• ใช้การตอบรับด้วยตนเองเพื่อให้ข้อความยังคงอยู่ในคิวจนกว่ากระบวนการของคุณจะเสร็จสิ้น
• กำหนดขีด จำกัด การดึงข้อมูลล่วงหน้าต่อคิวเป็น 1 ด้วยbasic.qos

หากคุณมี 4 คิว A, B, C และ D ซึ่งมีจำนวนข้อความที่แตกต่างกันเมื่อคุณเริ่มผู้บริโภค:

1. เมื่อคุณสมัครเป็นครั้งแรกขีด จำกัด การดึงข้อมูลล่วงหน้าจะหมายความว่าหนึ่งข้อความจากแต่ละคิวจะถูกส่งไปยังช่อง เรียกพวกเขาว่า A1, B1, C1 และ D1
2. ไลบรารีไคลเอ็นต์จะเพิ่มเหตุการณ์แบบอะซิงโครนัสในแอปพลิเคชันของคุณสำหรับแต่ละเหตุการณ์เหล่านี้
3. เธรดผู้ปฏิบัติงานคนเดียวของคุณจะจัดการเหตุการณ์แรกเหล่านี้และเริ่มประมวลผลข้อความ A1
4. จนกว่าคุณจะรับทราบข้อความนั้นด้วยตนเองจะไม่มีข้อความอื่นมาถึง
5. เมื่อคุณรับทราบข้อความแรก (A1) คุณสามารถดึงข้อความใหม่ล่วงหน้าจากคิวนั้นได้ (A2)
6. ในขณะเดียวกันเธรดผู้ปฏิบัติงานของคุณจะปลดบล็อกและจัดการกับเหตุการณ์ถัดไปที่ถูกเพิ่มขึ้นแล้วสำหรับข้อความ B1
7. เมื่อคุณประมวลผลเหตุการณ์ที่รอดำเนินการสำหรับ B1, C1 และ D1 เพียงครั้งเดียวเท่านั้นเธรดผู้ปฏิบัติงานจะเห็นเหตุการณ์สำหรับข้อความ A2
8. ตราบใดที่คิวยังมีข้อความรออยู่ระบบจะประมวลผลแบบปัดเศษ แม้ว่าคิวทั้งหมดจะว่างเปล่า แต่ก็จะกลับเข้าสู่การหมุนเวียนทันทีที่มีข้อความมาถึงเนื่องจากจะมีการดึงข้อความจากคิวที่ไม่ว่างไว้ล่วงหน้าเพียงข้อความเดียวส่วนที่เหลือจะรออยู่บนเซิร์ฟเวอร์ RabbitMQ