In this implementation, we'll focus on how to establish a RabbitMQ consumer in an Android application to receive messages from the queue. The following code demonstrates a robust message subscription mechanism with automatic reconnection capabilities.

Message Subscription Implementation

private void setupMessageReceiver(final Handler uiHandler) {
    // Create a dedicated thread for message consumption
    receiverThread = new Thread(() -> {
        while (shouldContinueReceiving) {
            try {
                // Establish connection to RabbitMQ server
                Connection rabbitConnection = connectionFactory.newConnection();
                Channel rabbitChannel = rabbitConnection.createChannel();
                
                // Configure quality of service
                rabbitChannel.basicQos(1);
                
                // Generate unique queue name
                String uniqueQueueName = "androidClient_" + System.currentTimeMillis();
                
                // Declare non-durable, exclusive, auto-delete queue
                AMQP.Queue.DeclareOk queueDeclaration = rabbitChannel.queueDeclare(
                    uniqueQueueName, false, false, true, null);
                
                // Bind queue to exchange with routing key
                rabbitChannel.queueBind(
                    queueDeclaration.getQueue(), 
                    "dataExchange", 
                    "dataRoutingKey");
                
                // Stop connection cycling after successful setup
                shouldContinueReceiving = false;
                
                // Create message consumer
                MessageConsumer consumer = new DefaultConsumer(rabbitChannel) {
                    @Override
                    public void handleDelivery(String consumerTag, 
                                             Envelope envelope, 
                                             AMQP.BasicProperties properties, 
                                             byte[] body) throws IOException {
                        
                        // Convert message payload to string
                        String messageContent = new String(body, "UTF-8);
                        
                        // Parse JSON message to data object
                        DataModel dataObject = jsonParser.fromJson(
                            messageContent, DataModel.class);
                        
                        // Prepare message for UI thread
                        android.os.Message uiMessage = uiHandler.obtainMessage();
                        Bundle dataBundle = new Bundle();
                        dataBundle.putParcelable("dataObject", dataObject);
                        uiMessage.setData(dataBundle);
                        
                        // Handle different message types
                        switch (dataObject.getEventType()) {
                            case "STATUS_UPDATE":
                                uiMessage.what = MSG_STATUS_UPDATE;
                                uiHandler.sendMessage(uiMessage);
                                break;
                                
                            case "ERROR_ALERT":
                                uiMessage.what = MSG_ERROR_ALERT;
                                uiHandler.sendMessage(uiMessage);
                                break;
                        }
                    }
                };
                
                // Start consuming messages
                rabbitChannel.basicConsume(queueDeclaration.getQueue(), true, consumer);
                
                // Notify UI of successful connection
                android.os.Message connectionStatus = new android.os.Message();
                connectionStatus.what = CONNECTION_SUCCESS;
                uiHandler.sendMessage(connectionStatus);
                
                // Monitor connection status
                while (!shouldContinueReceiving) {
                    if (!rabbitConnection.isOpen()) {
                        try {
                            Thread.sleep(RECONNECT_DELAY);
                            shouldContinueReceiving = true;
                        } catch (InterruptedException e) {
                            break;
                        }
                    }
                }
                
            } catch (Exception connectionException) {
                // Handle connection errors
                if (reconnectionAttempts > MAX_RECONNECT_ATTEMPTS) {
                    // Notify UI of connection failure
                    android.os.Message failureMessage = new android.os.Message();
                    Bundle errorBundle = new Bundle();
                    errorBundle.putString("error", "Reconnection attempts exhausted");
                    errorBundle.putBoolean("shouldClose", true);
                    failureMessage.setData(errorBundle);
                    uiHandler.sendMessage(failureMessage);
                    resetReconnectionCounter();
                    return;
                }
                
                // Notify UI of reconnection attempt
                android.os.Message retryMessage = new android.os.Message();
                Bundle retryBundle = new Bundle();
                retryBundle.putBoolean("shouldClose", false);
                retryBundle.putString("status", 
                    "Connection lost. Attempting reconnection (" + 
                    reconnectionAttempts + "/" + MAX_RECONNECT_ATTEMPTS + ")");
                retryMessage.setData(retryBundle);
                uiHandler.sendMessage(retryMessage);
                
                incrementReconnectionCounter();
                logConnectionError(connectionException);
                
                // Wait before retrying
                Thread.sleep(RECONNECT_DELAY);
            }
        }
    });
    
    // Start the receiver thread
    receiverThread.start();
}

Connection Cleanup Method

public void terminateConnection() {
    // Create thread for connection closing
    Thread shutdownThread = new Thread(() -> {
        try {
            // Safely close connection if it exists and is open
            if (rabbitConnection != null && rabbitConnection.isOpen()) {
                rabbitConnection.close();
            }
            
            // Safely close channel if it exists and is open
            if (rabbitChannel != null && rabbitChannel.isOpen()) {
                rabbitChannel.close();
            }
        } catch (IOException | TimeoutException closeException) {
            logError("Error while closing connection", closeException);
        }
    });
    
    // Start and interrupt the shutdown thread
    shutdownThread.start();
    shutdownThread.interrupt();
    
    // Stop other running threads
    if (senderThread != null) {
        senderThread.interrupt();
    }
    
    if (receiverThread != null) {
        receiverThread.interrupt();
    }
}

This implementation provides a complete solution for receiving messaegs from RabbitMQ in an Android application, with proper error handling, automatic reconnection, and thread management. The code separates concerns by using a dedicated thread for message consumption while communicating with the UI thread through a Handler.