rabbitmq AlreadyClosedException

Question

I am using the rabbitmq server with a java application. When the application receives a message on a particuliar queue, it generates some data and sends them on another queue.

When running the application, it receives the message, generates the data and sends them on the right queue. The data are well received on the server and they are correct. But when the application tries to send an acknowledgement to the server, I get an AlreadyClosedException.

I have the following message in the logs of the server: closing AMQP connection.

Here is the code of the handleDelivery function, in the rabbitMQ consumer class:

public void handleDelivery( String consumerTag, Envelope envelope, asicProperties properties, byte[] body )
        throws IOException {
    actionManager.receivedSelectedData( body );
    getChannel().basicAck( envelope.getDeliveryTag(), false );
}

Here is the code in the method receivedSelectedData(), where the data are generated before being sent:

public void receivedSelectedData( byte[] body ) {
    differentialEquations = differentialEquationsObjectManager.fromBytes( body );
    timeSeriesCollection.removeAllSeries();
    for ( int i = 0; i < differentialEquations.size(); i++ ) {
        differentialEquation = differentialEquations.get( i );
        for ( int j = 0; j < differentialEquation.getSystems().size(); j++ ) {
            try {
                generator = new NumericalMethod( differentialEquation, j );
            } catch ( Exception e ) {
                e.printStackTrace();
            }
            try {
                timeSeriesCollection.addSeries( generator.equationToTimeseriesRK4( 10.0 ) );
            } catch ( Exception e ) {
                e.printStackTrace();
            }
        }
    }
    sender.publish( RabbitMQGenerateSender.GENERATE_DATA_QUEUE_NAME,
            timeSeriesCollectionObjectMnanager.toBytes( timeSeriesCollection ) );
    }

The queue seems to be correctly declared, here is my queue declaration:

protected void declareQueue( String queueName ) {
        try {
            channel.queueDeclare( queueName, true, false, false, null );
        } catch ( IOException e ) {
            e.printStackTrace();
        }
}

and the channel declaration:

try {
            connection = factory.newConnection();
            channel = connection.createChannel();
            int prefetchCount = 1;
            channel.basicQos( prefetchCount );
        } catch ( IOException | TimeoutException e ) {
            e.printStackTrace();
        }

I have some other applications using rabbitmq with the same channel and queue declaration parameters and they work well. I only have one application that systematically fails when sending acknowledgment.

Here is the getChannel() method:

public Channel getChannel() {
    return channel;
}

How is the receiver queue is declared? Is it set to AutoDelete true? If so an Ack will throw an exception. — Mr Mush

Mr Mush Mr Mush · Accepted Answer · 2015-12-15T15:50:06

The receiver queue must be declared as AutoDelete = false if ACK functionality is to be supported. Here is an example in C# (there might be small differences than Java)

private bool PushDataToQueue(byte[] data, string queueName, ref string error)
{
    try
    {
        if (_connection == null || !_connection.IsOpen)
            _connection = _factory.CreateConnection();

        using (IModel channel = _connection.CreateModel())
        {
            if (AutoCloseConnection)
                _connection.AutoClose = AutoCloseConnection;
            // Set the AutoDelete as false, fourth parameter!!!
            channel.QueueDeclare(queueName, true, false, false, null);
            channel.BasicPublish("", queueName, null, data);
            if (!channel.IsClosed)
                channel.Close();
        }
    }
    catch (Exception e)
    {
        error = string.Format("Failed pushing data to queue name '{0}' on host '{1}' with user '{2}'. Error: {3}", queueName, _factory.HostName, _factory.UserName,
            e.GetCompleteDetails());
        return false;
    }

    return true;
}

rabbitmq AlreadyClosedException

1 Answers