RabbitMQ 玩 Java Akka

Question

我正在使用 Play Framework 2.2.2，并且正在使用 JavaAkka（Akka Actor System）实现 RabbitMQ 消费者应用程序。所以我有一个 MainActor，它在 Play 应用程序启动时使用 Global.OnStart 函数进行初始化。 MainActor 创建一个 RabbitMQ 通道，然后从队列开始消费。该队列中的每条消息都是另一个队列的名称，该队列必须分配给另一个必须从消息中提到的队列开始消费的子actor或子actor。所以本质上，我有一个订阅了一个 RabbitMQ 队列的 MainActor 和几个由 Main Actor 创建的子 Actor，每个子 Actor 都订阅了它们自己的 RabbitMQ 队列。问题是由于某种原因，我不能培养超过 7 个儿童演员。我怀疑是子角色中的 while(true) 构造等待来自 RabbitMQ 的消息。这是我的实现：

主要演员：

import play.Logger;
import com.typesafe.config.ConfigFactory;

import java.io.IOException;

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

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.QueueingConsumer;
import play.libs.Akka;
import util.RabbitMQConnection;

public class MainActor extends UntypedActor {

@Override
public void onReceive(Object msg) throws Exception {

        try{
            Connection connection = RabbitMQConnection.getConnection();
            Channel channel = connection.createChannel();

            String main_queue_name = ConfigFactory.load().getString("rabbitmq.default_queue");

            channel.queueDeclare(main_queue_name, false, false, false, null);

            QueueingConsumer consumer = new QueueingConsumer(channel);
            channel.basicConsume(main_queue_name, true, consumer);

            while (true) {

                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());

                System.out.println(" [x] Received '" + message + "'");

                ActorRef childActor = getContext().actorOf(Props.create(childActor.class));
                childActor.tell(message, getSelf());
            }
        }catch (Exception e){
            System.out.println(e.toString());
        }
    }
}

儿童演员：

import play.Logger;
import com.typesafe.config.ConfigFactory;
import java.io.IOException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectWriter;
import play.libs.Akka;
import play.libs.Json;

import akka.actor.UntypedActor;

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.QueueingConsumer;
import util.RabbitMQConnection;


public class childActor extends UntypedActor {

@Override
public void onReceive(Object msg) throws Exception {

    ObjectWriter ow = new ObjectMapper().writer().withDefaultPrettyPrinter();
    String queue_name = ow.writeValueAsString(msg);

    try{
        Connection connection = RabbitMQConnection.getConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare(queue_name, false, false, false, null);

        QueueingConsumer consumer = new QueueingConsumer(channel);
        channel.basicConsume(queue_name, true, consumer);

        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String message = new String(delivery.getBody());


            JsonNode jsonMsg = Json.parse(message);

            // Call some function to process the message

        }
    }catch (Exception e){
        System.out.println(e.toString());
    }
}
}

Answer 1

我认为在这种情况下您没有正确使用 Actor。在我看来，对于给定的演员，您不应该在 receive 方法中使用 while(true) 。此外，QueueingConsumer 已被弃用，rabbitmq 建议使用接口Consumer 或默认无操作实现DefaultConsumer 实现您的消费者。

我的做法是：

• 为rabbitmq 实现一个定制的消费者，它会在每次收到东西时向actor 发送一条消息。
• 为主要参与者使用该实现。将队列名称作为消息发送，并使用队列名称作为构造函数字段启动一个新的子 Actor。
• 为子演员使用该实现。将收到的消息发送给 actor，并在 actor 本身中进行 JSON 解析。

这里有一些代码：（警告：未编译或测试）

自定义rabbitmq消费者：

public class MyCustomRabbitMQConsumer extends DefaultConsumer {

    private ActorRef destinationActor;

    public MyCustomRabbitMQConsumer(ActorRef destinationActor) {
        this.destinationActor = destinationActor;
    }

    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
        destinationActor.tell(new String(body));
    }

}

主要演员：

import play.Logger;
import com.typesafe.config.ConfigFactory;

import java.io.IOException;

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

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.QueueingConsumer;
import play.libs.Akka;
import util.RabbitMQConnection;

public class MainActor extends UntypedActor {

    private MyCustomRabbitMQConsumer rabbitConsumer;

    @Override
    public void preStart() {
        Connection connection = RabbitMQConnection.getConnection();
        Channel channel = connection.createChannel();

        String main_queue_name = ConfigFactory.load().getString("rabbitmq.default_queue");
        channel.queueDeclare(main_queue_name, false, false, false, null);

        rabbitConsumer = new MyCustomRabbitMQConsumer(getSelf());
        channel.basicConsume(main_queue_name, true, rabbitConsumer);
    }

    @Override
    public void onReceive(Object msg) throws Exception {
        if(msg instanceOf String) {
            String queueName = (String) msg;
            System.out.println(" [x] Received '" + queueName + "'");
            getContext().actorOf(Props.create(childActor.class, queueName));
        }
    }
}

儿童演员：

import akka.actor.UntypedActor;

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.QueueingConsumer;
import util.RabbitMQConnection;


public class ChildActor extends UntypedActor {

    private MyCustomRabbitMQConsumer rabbitConsumer;
    private String queueName;

    public ChildActor(String queueName) {
        this.queueName = queueName;
    }

    @Override
    public void preStart() {
        Connection connection = RabbitMQConnection.getConnection();
        Channel channel = connection.createChannel();

        String main_queue_name = ConfigFactory.load().getString("rabbitmq.default_queue");
        channel.queueDeclare(queueName, false, false, false, null);

        rabbitConsumer = new MyCustomRabbitMQConsumer(getSelf());
        channel.basicConsume(queueName, true, rabbitConsumer);
    }


    @Override
    public void onReceive(Object msg) throws Exception {

        if(msg instanceOf String) {
            String strMsg = (String) msg;
            JsonNode jsonMsg = Json.parse(message);

            // Call some function to process the message
        }
    }
}

这应该适用于 n 个演员。