Akka演员Java竞争消费者

Question

我正在尝试使用Akka和Camel的竞争性事件消费者实现。我使用Akka 2.3.2和Camel 5.8.0。我正在将camel连接到ActiveMQ代理，并使用生产者从另一端生成消息。在以下代码中，EventManager是创建消费者池的主机，Event处理器是消息处理演员。

EventManager.java

    import org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.camel.component.ActiveMQComponent;
    import org.apache.camel.CamelContext;

    import akka.actor.ActorRef;
    import akka.actor.ActorSystem;
    import akka.actor.Props;
    import akka.actor.UntypedActor;

    import akka.camel.Camel;
    import akka.camel.CamelExtension;
    import akka.japi.Creator;
    import akka.routing.RoundRobinPool;

    public class EventManager {




        private final ActorSystem akkaSystem;

        private CamelContext camelContext = null;

        private ActorRef workRouter;

        public EventManager(ActorSystem system) {
            akkaSystem = system;

            initialize();
        }

        public void initialize() {

            Camel camel = CamelExtension.get(akkaSystem);

            camelContext = camel.context();

            ActiveMQComponent activemqComponent = ActiveMQComponent.activeMQComponent("tcp://localhost:61616");
            activemqComponent.setDeliveryPersistent(false);
            camelContext.addComponent("activemq",activemqComponent );


            int numOfWorkers = 5;

// distributing the message processing across a pool of 5 actors
            ActorRef workRouter =
                      akkaSystem.actorOf(new RoundRobinPool(numOfWorkers).props(Props.create(EventProcessor.class)), 
                        "workRouter");

        }

    }

EventProcessor.java

import org.apache.log4j.Logger;



import akka.actor.UntypedActor;
import akka.camel.CamelMessage;
import akka.camel.javaapi.UntypedConsumerActor;

public class EventProcessor extends UntypedConsumerActor{

    private static final Logger LOGGER = Logger.getLogger(EventProcessor.class);
    public EventProcessor() {

    }

    public void onReceive(Object message) {
        if(message instanceof CamelMessage) {
            CamelMessage camelMessage = (CamelMessage) message;
            String body = camelMessage.getBodyAs(String.class, getCamelContext());
                LOGGER.info("Message handled by :" +this.getSelf().path().name());

                LOGGER.info("Message body:" + body);
        }


    }

    public boolean autoAck() {
        return true;
    }
    public String getEndpointUri() {
        return "activemq:queue:dest";
    }

}

我看到的问题是，消息似乎被单个参与者使用，而没有在池中分发。我需要创建一个单独的骆驼路线来分发吗？我还希望将处理分布在不同的物理节点上。感谢您的意见和最佳实践。

Answer 1

尝试在AMQendpoint上设置并发使用者

return "activemq:queue:dest?concurrentConsumers=50";

..或在endpoint上启用异步使用者

return "activemq:queue:dest?asyncConsumer=true";

此外，您似乎弄错了版本信息。没有Camel版本5.8。我假设这是AMQ版本。