##Active MQ configuration
activemq.broker-url=(tcp://172.18.13.68:61616,tcp://172.18.13.69:61617)?ha=true&retryInterval=1000&reconnectAttempts=-1
activemq.user=admin
activemq.password=admin
activemq.topic=TestTopic
server.port=8084

Code:

@Service
public class TopicMessageConsumer implements MessageListener,ExceptionListener, InitializingBean, DisposableBean {


	@Value("${activemq.broker-url}")
	private String activeMqBrokerUri;

	@Value("${activemq.user}")
	private String username;

	@Value("${activemq.password}")
	private String password;

	@Value("${activemq.topic}")
	private String topicDestination;

	private Connection connection;

	private Session session;

	private long ms=0;

	public void run() throws JMSException {
		ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(activeMqBrokerUri,username, password);
		connection = factory.createConnection();
		connection.setClientID("clientID2");
		connection.setExceptionListener(this);
		connection.start();
		session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
		Topic destination = session.createTopic(topicDestination);
		MessageConsumer consumer = session.createDurableSubscriber(destination,"subsname");
		consumer.setMessageListener(this);
		System.out.println(String.format("TopicMessageConsumer Waiting for messages at %s %s", topicDestination, this.activeMqBrokerUri));
	}

	@Override
	public void onMessage(Message message) {
		String msg;
		try {
			msg = ((TextMessage) message).getText().replaceAll("#", "");
			System.out.println(new Date()+ " ["+ms+"] "+ msg);
			ms++;
		} catch (Exception e) {
			e.printStackTrace();
		}
	}


	@Override
	public void destroy() throws Exception {
		session.close();
		connection.close();
	}

	@Override
	public void afterPropertiesSet() throws Exception {
		run();
	}

	@Override
	public void onException(JMSException exception) {
		System.out.println("Excepcion---------------------");

	}

}