ActiveMQ Detect neue Verbraucher Thema

Question

Ich habe ein Standard activeMQ Broker

private static String localVMurl = "vm://localhost"; 

    broker = new BrokerService(); 
    broker.addConnector(localVMurl); 
    broker.start(); 

und alles ist gut. Was mein Ziel ist, dass ein Konsument sich mit dem Broker zu einem bestimmten Thema verbindet. Sobald diese Verbindung detectd ist, wird der Broker entweder Nachrichten weiterleiten, wenn ein Producer aktiv an dem Thema produziert, ODER der Broker wird einen neuen Producer für dieses spezielle Thema starten. Um dies zu tun, muss ich jedoch feststellen, dass ein neuer Verbraucher ein bestimmtes Thema anmeldet und anfordert.

meine grundlegenden Verbraucher Code:

ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(remoterURL); 
    Connection connection = connectionFactory.createConnection(); 
    connection.setClientID("clinet1"); 
    connection.start(); 

    session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); 
    Topic topic = session.createTopic("some_topic"); 
    MessageConsumer consumer = session.createConsumer(topic); 
    consumer.setMessageListener(new MyMessageListener()); 

und ich kann in den Broker-Logs sehen:

<161005 11:03:41> [.0.1:64433@5001] DEBUG tRegion - localhost adding destination: topic://ActiveMQ.Advisory.Consumer.Topic.some_topic 

Ich weiß also, der Verbraucher verbindet, und abonniert diesem Thema, ich muss nur fangen dieses Ereignis irgendwie.

Irgendwelche Gedanken, wie man das macht?

Answer 1

Advisory Message ist, was Sie brauchen. Jedes Mal, wenn Sie eine Nachricht mit diesem Code erhalten, bedeutet dies, dass Sie einen neuen Konsumenten starten oder stoppen.

doc http://activemq.apache.org/advisory-message.html

Beispiel:

//org.apache.activemq.advisory.AdvisorySupport.getDestinationAdvisoryTopic(Destination) 
Destination advisoryDestination = AdvisorySupport.getConsumerAdvisoryTopic(topic) 
MessageConsumer consumer = session.createConsumer(advisoryDestination); 
consumer.setMessageListener(this); 

public void onMessage(Message msg){ 
    if (msg instanceof ActiveMQMessage){ 
     try { 
      ActiveMQMessage aMsg = (ActiveMQMessage)msg; 
      ConsumerInfo consumer = (ConsumerInfo) aMsg.getDataStructure(); 
     } catch (JMSException e) { 
      log.error("Failed to process message: " + msg); 
     } 
    } 
}