使用 RabbitMqBundle 的管道和过滤器

Question

我正在尝试使用 RabbitMqBundle 创建管道和过滤器 (http://www.eaipatterns.com/PipesAndFilters.html) 实现。最终设置应如下所示：

[main_topic_exchange]

   |---BK=S1.K1---> consumer1.1 ---BK=S2.K1---> consumer2.1---> ...
   |---BK=S1.K2---> consumer1.2 ---BK=S2.K2---> consumer2.2---> ...
                             ...............
   |---BK=S1.Kn---> consumer1.n ---BK=S2.Kn---> consumer2.n---> ...

基本上，我需要能够配置管道“蓝图”并根据需要启动尽可能多的管道实例（以实现可扩展性）。

• 每个管道或管道中的第一个使用者都绑定到一个交换器（例如 BK=S1.k1）
• 消息从一个计算阶段（例如 BK=S2.K1）传递到下一个计算阶段
• 每个队列都是独占的
• 必须保证通过管道的消息顺序
• 进程将由 supervisord 管理

我的解决方案：

1. 使用现有的包并编码生产者和直接在回调中指示下一阶段的路由键 - 不太灵活，硬编码

2. 扩展 rabbitmqbundle 并创建具有路由密钥设置器方法的消费者并从命令行传递它：

./app/console rabbitmq:pipe1_consumer1 --route=S1.K1 --next=S2.K1 ./app/console rabbitmq:pipe1_consumer2 --route=S2.K1 --next=S3.K1 ./app/console rabbitmq:pipe1_consumer2 --route=S2.K1 --next=S3.K1

--next 将指示下一个计算阶段（消费者）的路由密钥。

我不确定最优雅的解决方案是什么，有什么想法吗？

谢谢。

Answer 1

RabbitMQBundle 作者在这里。该捆绑包的最初目标是实现几种消息传递模式。考虑到管道和过滤器只是另一种模式，最好的办法是实现它并将其添加到库中。我会欢迎这样的补充。