Oldsound rabbitmq 捆绑多个消费者配置

Question

我正在尝试使用 https://github.com/php-amqplib/RabbitMqBundle 和 Symfony2 框架来实现 RabbitMQ。

我已经设法让 1 个生产者和 1 个消费者一起工作，但问题是当我使用多个消费者时。

这是我的配置：

old_sound_rabbit_mq:
    connections:
        default:
            host:     'localhost'
            port:     5672
            user:     'guest'
            password: 'guest'
            vhost:    '/'
            lazy:     false
            connection_timeout: 3
            read_write_timeout: 3

            # requires php-amqplib v2.4.1+ and PHP5.4+
            keepalive: false

            # requires php-amqplib v2.4.1+
            heartbeat: 0

            #requires php_sockets.dll
#            use_socket: true # default false
    producers:
        soccer_team_stat:
            connection:       default
            exchange_options: {name: 'soccer_team_stat_ex', type: direct}
            queue_options:    {name: 'soccer_team_stat_qu'}
        soccer_team_stat_form:
            connection: default
            exchange_options: {name: 'soccer_team_stat_ex', type: direct}
            queue_options:    {name: 'soccer_team_stat_form_qu'}
    consumers:
        soccer_team_stat:
            connection:       default
            exchange_options: {name: 'soccer_team_stat_ex', type: direct}
            queue_options:    {name: 'soccer_team_stat_qu'}
            callback:         myapp.soccer_team_stat.consume
        soccer_team_stat_form:
            connection:       default
            exchange_options: {name: 'soccer_team_stat_ex', type: direct}
            queue_options:    {name: 'soccer_team_stat_form_qu'}
            callback:         myapp.soccer_team_stat_form.consume

服务定义：

  <services>

        <service class="MyApp\EtlBundle\Producers\SoccerTeamStatProducer" id="myapp.soccer_team_stat.produce">
            <argument type="service" id="old_sound_rabbit_mq.soccer_team_stat_producer"/>
        </service>

        <service class="MyApp\EtlBundle\Producers\SoccerTeamStatProducer" id="myapp.soccer_team_stat_form.produce">
            <argument type="service" id="old_sound_rabbit_mq.soccer_team_stat_producer"/>
        </service>

        <service class="MyApp\EtlBundle\Consumers\SoccerTeamStatConsumer" id="myapp.soccer_team_stat.consume">
            <argument type="service" id="service_container"/>
        </service>

        <service class="MyApp\EtlBundle\Consumers\SoccerTeamStatFormConsumer" id="myapp.soccer_team_stat_form.consume">
            <argument type="service" id="service_container"/>
        </service>

    </services>

在 php app/console rabbitmq:consumer -d football_team_stat_form 我得到：

[Symfony\Component\DependencyInjection\Exception\ServiceNotFoundException] 您请求的服务不存在 “old_sound_rabbit_mq.soccer_team_stat_form_consumer”。

我尝试了各种组合，包括使用 multiple_consumers 配置键，但没有成功。我错过了什么？

Answer 1

如果routing_key 和binding_key 均未设置，direct 交换的行为将类似于fanout 并将消息发送到它知道的所有队列，因此根据我从您的配置中看到的情况，您是最好使用fanout，如下所示。

old_sound_rabbit_mq:
    connections:
        default:
            host:     %rabbit_mq_host%
            port:     %rabbit_mq_port%
            user:     %rabbit_mq_user%
            password: %rabbit_mq_pswd%
            vhost:    /
            lazy:     true
    producers:
        soccer_team_stat:
            connection:       default
            exchange_options: { name: 'soccer_team_stat_ex', type: fanout }
        soccer_team_stat_form:
            connection:       default
            exchange_options: { name: 'soccer_team_stat_form_ex', type: fanout }
    consumers:
        soccer_team_stat:
            connection:       default
            exchange_options: { name: 'soccer_team_stat_ex', type: fanout }
            queue_options:    { name: 'soccer_team_stat_qu' }
            callback:         myapp.soccer_team_stat.consume
        soccer_team_stat_form:
            connection:       default
            exchange_options: { name: 'soccer_team_stat_form_ex', type: fanout }
            queue_options:    { name: 'soccer_team_stat_form_qu' }
            callback:         myapp.soccer_team_stat_form.consume

这个RabbitMQ fanout example with symfony including 2 Producer & 2 Exchange & 2 Queue & N Worker & 2 Consumer 是一个完整的例子（实际上是你的问题的完整答案/你想要做什么的已经制作的版本），它显示了事情是如何在 symfony 应用程序中完成的。我建议您遵循那里使用的模式。非常容易遵循和维护。如果您想要更多示例，只需在该博客中搜索 RabbitMQ 关键字即可。