带有AMQP的Android

Question

我正在尝试在我的Android应用程序中使用AMQP消息队列。我首先在java应用程序中测试了代码，没有问题，但是当我将相同的代码放入Android应用程序中时，没有抛出错误，但是AMQP通道在收到第一条消息后意外关闭。我尝试向通道添加一个关闭监听程序，但没有返回任何信息。谁能帮我弄清楚Rabbitmq频道为什么要关闭？

// ------------------------------------------------------------------------
// Function to make AMQP connection and subscribe
// ------------------------------------------------------------------------
int connAmqp()
{
    factory = new ConnectionFactory();
    factory.setHost(PI_AMQP_BROKER_URL);
    factory.setUsername(AMQP_BROKER_USERNAME);
    factory.setPassword(AMQP_BROKER_PASSWORD);
    factory.setPort(AMQP_BROKER_PORT);
    connection=null;
    connectionStatus = AMQP_CONNECTING;
    try {

        connection = factory.newConnection();
        Log.i("log_amqp_conn","Successfully connected to AMQP broker");
        connectionStatus = AMQP_CONNECTED;
        channel = connection.createChannel();

        channel.addShutdownListener(new ShutdownListener() {
            @Override
            public void shutdownCompleted(ShutdownSignalException cause) {
                Log.w("log_amqp_shutdown",cause.getCause().toString()); 
            }
        });

        channel.exchangeDeclare(AMQP_EXCHANGE_NAME, "topic");
        Log.i("log_amqp_conn","Successfully connected to Exchange: " + AMQP_EXCHANGE_NAME);
        channel.queueDeclare(RX_QUEUE_NAME, false, false, false, null);
        String queueBind = RX_BINDING;
        Log.i("log_amqp_conn","Successfully declared queue: " + RX_QUEUE_NAME);
        channel.queueBind(RX_QUEUE_NAME, AMQP_EXCHANGE_NAME, queueBind);
        Log.i("log_amqp_conn","Successfully binding to: " + queueBind);
    } catch (TimeoutException e) {
        Log.i("log_amqp_conn","Connection timeount - Failed to connect to AMQP broker");
        Log.i("log_amqp_conn",e.toString());
        connectionStatus = AMQP_NOTCONNECTED_TIMEOUT;
    } catch (IOException e) {
        Log.i("log_amqp_conn","Failed to connect to AMQP broker");
        Log.i("log_amqp_conn",e.toString());
        connectionStatus = AMQP_NOTCONNECTED_UNKNOWNREASON;
    }catch (Exception e) {
        Log.i("log_amqp_conn","Failed to connect to AMQP broker");
        Log.i("log_amqp_conn",e.toString());
        connectionStatus = AMQP_NOTCONNECTED_UNKNOWNREASON;
    }

    Log.i("log_amqp_conn"," [*] Waiting for messages.");

    consumer = new DefaultConsumer(channel) {
        @Override
        public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
                throws IOException {
            Log.i("log_amqp_conn", "handleDelivery() called");
            String message = new String(body, "UTF-8");
            Log.i("log_amqp_conn"," [Rx] Received: " + message);

            //DO SOME WORK HERE

        }
     };
    try {
        Log.i("log_amqp_conn", "basicConsume() called");
        channel.basicConsume(RX_QUEUE_NAME, true, consumer);
    } catch (IOException e) {
        Log.e("log_amqp_consume",e.toString());
    }

    return 0;
}

Answer 1

在花了很多时间查看错误的位置后，我发现由于“//在这里做一些工作”代码中的handleDelivery()函数中的一个未处理的异常，通道正在关闭。谢谢。