java - RabbitMQ 玩 Java Akka

Question

我正在使用 Play Framework 2.2.2，并且正在使用 JavaAkka(Akka Actor System)实现 RabbitMQ 消费者应用程序。因此，我有一个 MainActor，当 Play 应用程序使用 Global.OnStart 函数启动时，它会被初始化。 MainActor 创建一个 RabbitMQ channel ，然后从队列开始消费。该队列中的每条消息都是另一个队列的名称，该队列必须分配给另一个子参与者或子参与者，该子参与者必须开始从消息中提到的队列消费。所以本质上，我有一个 MainActor 订阅了一个 RabbitMQ 队列，还有几个由 Main Actor 创建的子 Actor，每个子 Actor 都订阅了自己的 RabbitMQ 队列。问题是，出于某种原因，我不能带出超过 7 个 child Actor 。我怀疑是子 actor 中的 while(true) 构造等待来自 RabbitMQ 的消息。这是我的实现:

主要 Actor :

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());
        }
    }
}

child Actor :

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 人员建议使用接口(interface) Consumer 或默认的无操作实现 DefaultConsumer 来实现消费者。

我的做法是:

• 为rabbitmq实现一个定制的消费者，每次它收到东西时都会向参与者发送一条消息。
• 为主角使用该实现。将队列名称作为消息发送，并使用队列名称作为构造函数字段启动一个新的子 Actor。
• 将该实现用于 child 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));
    }

}

主要 Actor :

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));
        }
    }
}

child Actor :

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 个 Actor 。