JMX: Building a JMX Notification Hub

Introduction

However, it's necessary for clients to individually subscribe to each MBean that they are interested in receiving notifications from. When there are many such MBeans, or when the MBeans may be created dynamically, this can pose a problem for the client.

This article shows how to build a JMX Notification Hub MBean which provides a single place for clients to subscribe to receive notifications from many MBeans.

JMX 1.2 Notifications - background

Whilst there is a publisher and a subscriber, this doesn't really follow the publish/subscribe paradigm. The publisher and the subscriber need to know about each other, whereas in a true publish/subscribe paradigm they are dissociated from each other. This leads to several limitations in the existing implementation, amongst which are:

1. The client must subscribe to each and every MBean it is interested in, even if there are thousands of them.
2. The client cannot pre-subscribe to an MBean before it is created, so must listen for creation notifications and deal with asynchronous subscription.
3. The MBean emitting the notification must maintain state for each client interested in receiving the notification, which otherwise it would not need to maintain.
4. If the client disconnects and reconnects, it must resubscribe to all MBeans, and may have lost notifications during the interval.

Building a JMX 1.2 Notification Hub

It's actually quite easy to build a simple one of these... all you need to do is write a NotificationHubMBean that is itself a NotificationEmitter, and which sits there listening to every other MBean and forwarding the Notifications.

Here's the basic MBean interface for our NotificationHub... we don't actually need anything at all in this interface, what's important for the MBean is that the implementation class declares itself as implementing NotificationEmitter:

public interface NotificationHubMBean { /** * accessor * @return number of notifications hub has received */ int getNotificationCount(); }

 

And here's the implementation of the NotificationHubMBean. All it does is subscribe to every other MBean in sight and forward their notifications.

import java.util.Set; import javax.management.MBeanRegistration; import javax.management.MBeanServer; import javax.management.MBeanServerNotification; import javax.management.Notification; import javax.management.NotificationBroadcasterSupport; import javax.management.NotificationListener; import javax.management.ObjectName; /** * A NotificationHubMBean implementation * that subscribes to all MBeans and forwards * their Notifications */ public class NotificationHub extends NotificationBroadcasterSupport implements NotificationHubMBean, MBeanRegistration, NotificationEmitter, NotificationListener { private MBeanServer mbs; private ObjectName myObjectName; /** private count of number of notifications we receive */ private int notificationCount; /** Creates a new instance of NotificationHub */ public NotificationHub() { } /** * accessor * @return number of notifications hub has received */ public int getNotificationCount() { return notificationCount; } /* MBeanRegistration Interface implementation */ /** * Implementation private. Does a queryNames() and tries to * subscribe to every MBean. * @param mbs * @param name * @throws java.lang.Exception * @return the ObjectName for this MBean */ public ObjectName preRegister(MBeanServer mbs, ObjectName name) throws Exception { this.mbs = mbs; this.myObjectName = name; // Subscribe to all existing ObjectNames that are broadcasters Set<ObjectName> objectNameSet = mbs.queryNames(new ObjectName("*:*"), null); for (ObjectName on : objectNameSet) { // We don't subscribe to any other NotificationHubs // as this could lead to an infinite loop! try { if (!mbs.isInstanceOf(on, this.getClass().getName())) { mbs.addNotificationListener(on, this, null, null); } } catch (Exception e) { // Ignore exception as the current MBean in the list // might not be a notification broadcaster. } } return name; } /** * implementation private. */ public void postRegister(Boolean registrationDone) { } /** * Implementation private. * Unregister our listener from all MBeans */ public void preDeregister() throws Exception { // Unsubscribe to all existing ObjectNames that are broadcasters Set<ObjectName> objectNameSet = mbs.queryNames(new ObjectName("*:*"), null); for (ObjectName on : objectNameSet) { // Try and unsubscribe from all mbeans try { mbs.removeNotificationListener(on, this); } catch (Exception e) { // Ignore exception as the current MBean in the list // might not be a notification broadcaster. } } } /** * implementation private */ public void postDeregister() { } //-------------------------------- // NOTIFICATION LISTENER INTERFACE //-------------------------------- /** * Receives notifications and forwards to all listeners. * It also manages notifications coming from the MBeanServerDelegate and * subscribes to new mbeans based upon REGISTRATION_NOTIFICATION events */ public void handleNotification(Notification notification, Object handback) { // If new MBeans are created then put listeners on them. String type = notification.getType(); notificationCount++; if (type != null && type.equals(MBeanServerNotification.REGISTRATION_NOTIFICATION)) { try { ObjectName on = ((MBeanServerNotification)notification).getMBeanName(); // Avoid loops by not subscribing to other Hubs. if (!mbs.isInstanceOf(on, this.getClass().getName())) { mbs.addNotificationListener(on, this, null, null); } } catch (Exception e) { // Ignore exception as the new created MBean might // not be a notification broadcaster. } } // Deliver all notifications to anyone subscribed to us sendNotification(notification); } }

 

An example program

import java.lang.management.ManagementFactory;
import java.util.Date;
import javax.management.MBeanServer;
import javax.management.Notification;
import javax.management.NotificationListener;
import javax.management.ObjectName;
import javax.management.timer.Timer;

/**
 * NotificationHub example application
 */
public class NotificationHubExample implements NotificationListener {

    public NotificationHubExample() {
        try {
            ObjectName on;

            on = new ObjectName("example:type=NotificationHubMBean");

            NotificationHubMBean nh = new NotificationHub();

            MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();

            mbs.registerMBean(nh, on);

            // Now add ourselves as a listener to the hub
            mbs.addNotificationListener(on, this, null, null);

            // Create a few timer mbeans that send regular notifications
            for (int i = 1 ; i < 4; i++) {
                on = new ObjectName("example:type=TimerMBean,name="+i);
                Timer timer = new Timer();
                timer.addNotification("TICK"+i, "tick from "+i,
                        null,
                        new Date(),
                        1000 * i);
                mbs.registerMBean(timer, on);
                timer.start();
            }
        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }

    /* from NotificationListener interface */
    public void handleNotification(Notification notification, Object handback) {
        System.out.println("Got notification : "+notification);
    }

    /**
     * @param args the command line arguments
     */
    public static void main(String[] args) {

        NotificationHubExample me = new NotificationHubExample();

        // wait for a keypress and exit
        try {
            int c = System.in.read();
        } catch (Exception ex) {
            ex.printStackTrace();
        }
        // Don't bother killing Timer threads, just exit
        System.exit(0);
    }
}

Got notification : javax.management.MBeanServerNotification[source=JMImplementation:type=MBeanServerDelegate][type=JMX.mbean.registered][message=]
Got notification : javax.management.MBeanServerNotification[source=JMImplementation:type=MBeanServerDelegate][type=JMX.mbean.registered][message=]
Got notification : javax.management.timer.TimerNotification[source=example:type=TimerMBean,name=1][type=TICK1][message=tick from 1]
Got notification : javax.management.MBeanServerNotification[source=JMImplementation:type=MBeanServerDelegate][type=JMX.mbean.registered][message=]
Got notification : javax.management.timer.TimerNotification[source=example:type=TimerMBean,name=2][type=TICK2][message=tick from 2]
Got notification : javax.management.timer.TimerNotification[source=example:type=TimerMBean,name=1][type=TICK1][message=tick from 1]
Got notification : javax.management.timer.TimerNotification[source=example:type=TimerMBean,name=1][type=TICK1][message=tick from 1]
Got notification : javax.management.timer.TimerNotification[source=example:type=TimerMBean,name=2][type=TICK2][message=tick from 2]
Got notification : javax.management.timer.TimerNotification[source=example:type=TimerMBean,name=3][type=TICK3][message=tick from 3]
Got notification : javax.management.timer.TimerNotification[source=example:type=TimerMBean,name=1][type=TICK1][message=tick from 1]
Got notification : javax.management.timer.TimerNotification[source=example:type=TimerMBean,name=1][type=TICK1][message=tick from 1]
Got notification : javax.management.timer.TimerNotification[source=example:type=TimerMBean,name=2][type=TICK2][message=tick from 2]
Got notification : javax.management.timer.TimerNotification[source=example:type=TimerMBean,name=1][type=TICK1][message=tick from 1]
Got notification : javax.management.timer.TimerNotification[source=example:type=TimerMBean,name=3][type=TICK3][message=tick from 3]
Got notification : javax.management.timer.TimerNotification[source=example:type=TimerMBean,name=1][type=TICK1][message=tick from 1]
Got notification : javax.management.timer.TimerNotification[source=example:type=TimerMBean,name=2][type=TICK2][message=tick from 2]
Got notification : javax.management.timer.TimerNotification[source=example:type=TimerMBean,name=1][type=TICK1][message=tick from 1]
Got notification : javax.management.timer.TimerNotification[source=example:type=TimerMBean,name=1][type=TICK1][message=tick from 1]
Got notification : javax.management.timer.TimerNotification[source=example:type=TimerMBean,name=2][type=TICK2][message=tick from 2] 

Limitations of this implementation

• Thread safety - there's a race condition when incrementing the notification count and another between the MBeanServer query and adding/removing ourselves as a listener to/from the MBeanServerDelegate
• Inability for clients to filter on an ObjectName pattern - so all Notifications get forwarded over the wire to the remote clients, rather than just those that they may be interested in ... this could be added by providing a special NotificationFilter implementation that clients could push that allows an ObjectName pattern to be used as the filter (but would require the client to have this on its classpath).
• Scalability - it might be preferable to be able to provide a NotificationHubMBean to work for a particular range of MBeans as defined by an ObjectName pattern, rather than one for all MBeans
• Additional features such as logging - one can use this centralized hub to facilitate debugging by writing out a fine-grained log record for each notification received using the Java logging API

JMX 2.0's Event Service

( Nov 10 2006, 04:41:35 PM CET ) Permalink