ActiveMQ connection event notification: Monitor broker status using Advisory Topics

This article details how to use the Advisory Topics function of Apache ActiveMQ to enable Java applications to monitor the connection creation, closing and other key events of the ActiveMQ broker in real time. By subscribing to a specific Advisory Topic, developers can receive and process notifications such as connection status changes, consumer/producer activities, temporary destination life cycles, etc., thereby achieving more granular monitoring and management of the message broker.

When building a distributed system based on message queues, it is crucial to understand the internal operating status of the message broker (Broker). In particular, monitoring the creation and closing of client connections can help us better understand system load, diagnose connection problems, or trigger specific business logic. Apache ActiveMQ provides a powerful mechanism to achieve this - Advisory Topics .

What are ActiveMQ Advisory Topics?

Advisory Topics are a special topic (Topic) built into ActiveMQ. The agent will publish notification messages about its internal events to these topics. By subscribing to these Advisory Topics, client applications can receive and process various agent events in real time without directly querying the agent status. These events cover all aspects from the life cycle of the client connection to the flow of messages.

Incident types covered by Advisory Topics

Advisory Topics can publish many types of event notifications, including but not limited to:

• Connections : The creation and closing of client connections.
• Consumers : Online and offline consumers.
• Producers : Producers are online and offline.
• Temporary Destinations : The creation and destruction of temporary queues or topics.
• Messages Expiring : Messages in the queue or topic have expired.
• No Consumers on Destination : Messages are sent to destinations with no active consumers.

This article will focus on how to monitor connection creation and closing events.

Monitor connection events

To monitor the connection creation and closing events of the ActiveMQ broker, we need to subscribe to the Advisory Topic named ActiveMQ.Advisory.Connection. The broker sends a notification message to this topic when a new client connects to the broker or when an existing connection is lost.

Sample code: Subscription connection Advisory Topic

The following Java code example demonstrates how to use the JMS API to subscribe to the ActiveMQ.Advisory.Connection topic and listen for connection events:

 import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQMessage; // Used to access ActiveMQ-specific message properties import javax.jms.*;

public class ActiveMQConnectionMonitor {

    // URL of ActiveMQ broker
    private static final String BROKER_URL = "tcp://localhost:61616";
    //Advisory Topic name for listening to connection events private static final String ADVISORY_CONNECTION_TOPIC = "ActiveMQ.Advisory.Connection";

    public static void main(String[] args) {
        Connection connection = null;
        Session session = null;
        MessageConsumer consumer = null;

        try {
            // 1. Create a JMS connection factory ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(BROKER_URL);

            // 2. Create and start the JMS connection connection = connectionFactory.createConnection();
            connection.start();

            // 3. Create a JMS session (non-transactional, automatically acknowledge messages)
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            // 4. Create Advisory Topic object Topic advisoryTopic = session.createTopic(ADVISORY_CONNECTION_TOPIC);

            // 5. Create a message consumer for subscribing to Advisory Topic
            consumer = session.createConsumer(advisoryTopic);

            // 6. Set the message listener and process the Advisory message when receiving it consumer.setMessageListener(new MessageListener() {
                @Override
                public void onMessage(Message message) {
                    try {
                        // Advisory messages are usually of type ActiveMQMessage and contain specific attributes if (message instanceof ActiveMQMessage) {
                            ActiveMQMessage amqMessage = (ActiveMQMessage) message;

                            //Extract event information from message properties String advisoryMessageType = amqMessage.getStringProperty("AdvisoryMessageType");
                            String connectionId = amqMessage.getStringProperty("connectionId");
                            boolean isConnectionStart = amqMessage.getBooleanProperty("isConnectionStart");
                            boolean isConnectionStop = amqMessage.getBooleanProperty("isConnectionStop");

                            System.out.println("----------------------------------------");
                            System.out.println("ActiveMQ connection event notification received:");
                            System.out.println(" Message ID: " amqMessage.getJMSMessageID());
                            System.out.println(" Advisory Message Type: " advisoryMessageType);
                            System.out.println("Associated connection ID: " connectionId);
                            System.out.println("Is it a connection start event? " isConnectionStart);
                            System.out.println("Is the connection stop event? " isConnectionStop);
                            System.out.println("----------------------------------------");

                            // Execute corresponding business logic according to event type if (isConnectionStart) {
                                System.out.println(" -> New connection established: " connectionId);
                                // For example: logging, updating connection status dashboard, etc.} else if (isConnectionStop) {
                                System.out.println(" -> Connection closed: " connectionId);
                                // For example: clean up resources, send alarms, etc.}

                        } else {
                            System.out.println("Received non-ActiveMQMessage type message: " message.getClass().getName());
                        }
                    } catch (JMSException e) {
                        System.err.println("An error occurred while processing the Advisory message: " e.getMessage());
                        e.printStackTrace();
                    }
                }
            });

            System.out.println("Listening to ActiveMQ connection events... Please start or close other ActiveMQ clients for testing.");
            System.out.println("Press Ctrl C or close the program to exit monitoring.");

            //Keep the main thread running so that the listener can continue to receive messages Thread.sleep(Long.MAX_VALUE);

        } catch (JMSException e) {
            System.err.println("JMS operation failed: " e.getMessage());
            e.printStackTrace();
        } catch (InterruptedException e) {
            System.err.println("Listening thread interrupt: " e.getMessage());
            e.printStackTrace();
        } finally {
            // 7. Close JMS resources and release the connection try {
                if (consumer != null) consumer.close();
                if (session != null) session.close();
                if (connection != null) connection.close();
                System.out.println("JMS resource has been closed.");
            } catch (JMSException e) {
                System.err.println("Failed to close JMS resource: " e.getMessage());
                e.printStackTrace();
            }
        }
    }
}

Maven dependency

To use the above code in a Java project, you need to add the Maven dependency of the ActiveMQ client library:

 <dependency>
    <groupid>org.apache.activemq</groupid>
    <artifactid>activemq-client</artifactid>
    <version>5.18.3</version> <!-- Use the client version corresponding to your ActiveMQ version -->
</dependency>

Interpretation of message attributes

Advisory messages are standard JMS messages, but ActiveMQ adds some specific attributes to describe the event. For connection advisory messages, the following properties are particularly useful:

• AdvisoryMessageType: Usually connection, indicating that this is a connection-related Advisory message.
• connectionId: The unique identifier of the client connection where the event occurred.
• isConnectionStart: Boolean value, if true, indicates that the connection has been established.
• isConnectionStop: Boolean value, if true, indicates that the connection has been closed.

By examining these properties, applications can accurately determine the type of event and take appropriate action.

Things to note

1. Performance impact : Although Advisory Topics are very useful, if there is very high client activity on the agent, generating a large number of Advisory messages may have a slight impact on the performance of the agent. In a production environment, whether to enable all Advisory events should be weighed based on actual needs.
2. Message persistence : Messages for Advisory Topics are usually non-persistent. This means that if a subscriber is not online when the event occurs, it will miss these notifications. If you need persistent Advisory notifications, you may need to configure a proxy or use other mechanisms.
3. Security : By default, any client can subscribe to Advisory Topics. In a production environment, you should consider configuring ActiveMQ security policies to restrict which users or applications can subscribe to these sensitive internal events.
4. Message structure : The specific attributes and content of the Advisory message may change as the ActiveMQ version is updated. It is recommended to consult the official documentation of the corresponding version of ActiveMQ to obtain the most accurate information.

Summarize

ActiveMQ Advisory Topics provide a powerful and flexible mechanism for Java applications to monitor internal events of the message broker. By subscribing to the ActiveMQ.Advisory.Connection topic, developers can easily implement real-time notifications of client connection creation and closure, thereby enhancing system observability, simplifying troubleshooting, and supporting smarter automated management. Mastering the use of Advisory Topics is a key step to deeply understand and effectively manage ActiveMQ brokers.

The above is the detailed content of ActiveMQ connection event notification: Monitor broker status using Advisory Topics. For more information, please follow other related articles on the PHP Chinese website!