Я использую Apache ActiveMQ 5.5 и у меня следующий сценарий
-Встроенный брокер
- Непостоянный производитель и тема
- потребитель темы.
Нормальные вещи работают - я публикую в теме, и подписчик использует из нее.
Я реализовал MessageListener, поэтому, когда потребитель подписывается и отписывается, я что-то печатаю, чтобы указать это.
Когда мне нужно закрыть потребителя, я просто вызываю метод close.
Иногда потребитель успешно закрывается - я вижу, что журнал и использование памяти в порядке.
Но иногда он не закрывается, хотя я вижу в журнале, что я вызвал метод close.
На этот раз в журнале MessageListener не упоминается, как подписчик отписался.
И в результате, использование памяти только увеличивается, потому что теперь издатель отправляет сообщения в тему, а я закрыл потребителя (который на самом деле не закрыт)
и перестал обрабатывать сообщения.
Так что я не уверен, где и как устранить эту проблему ......
Я думаю, что это как-то связано с рабочими потоками asynch activemq и их поведением.
Ниже приведены все классы, связанные с ActiveMQ, которые я использую ... дайте мне знать, если я должен выпустить дополнительный код.
public class Consumer {
private MessageConsumer consumer;
public Consumer(String brokerUrl, String topic, MessageListener ml) throws JMSException {
ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory(brokerUrl);
//with straight through processing of messages
//and optimized acknowledgement
cf.setAlwaysSessionAsync(false);
cf.setOptimizeAcknowledge(true);
Connection connection = cf.createConnection();
connection.start();
//-- Use the default session
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//-- Set the prefetch size for topics - by parsing a configuration parameter in
// the name of the topic
//-- topic=test.topic?consumer.prefetchSize=32766
Topic topicObj = session.createTopic(topic);
consumer = session.createConsumer(topicObj);
//-- Attach the passed in message listener
consumer.setMessageListener(ml);
}
/**
* @return the consumer
*/
public MessageConsumer getConsumer() {
return consumer;
}
}
public class ConsumerAdvisoryListener implements MessageListener {
private XMLLogUtil xlu;
private MyLogger ml;
public ConsumerAdvisoryListener() throws IOException{
xlu=XMLLogUtil.getInstance();
ml=xlu.getCustomLogger(ConsumerAdvisoryListener.class);
}
public void onMessage(Message message) {
ActiveMQMessage msg = (ActiveMQMessage) message;
DataStructure ds = msg.getDataStructure();
if (ds != null) {
switch (ds.getDataStructureType()) {
case CommandTypes.CONSUMER_INFO:
ConsumerInfo consumerInfo = (ConsumerInfo) ds;
ml.info("Consumer '" + consumerInfo.getConsumerId()
+ "' subscribed to '" + consumerInfo.getDestination()
+ "'");
break;
case CommandTypes.REMOVE_INFO:
RemoveInfo removeInfo = (RemoveInfo) ds;
ConsumerId consumerId = ((ConsumerId) removeInfo.getObjectId());
ml.info("Consumer '" + consumerId + "' unsubscribed");
break;
default:
ml.info("Unkown data structure type");
}
} else {
ml.info("No data structure provided");
}
}
}
public class EmbeddedBroker {
/**
* Singleton
*/
private static EmbeddedBroker INSTANCE;
private BrokerService broker;
/**
* Return singleton instance
* @return
*/
public static EmbeddedBroker getInstance(){
if(EmbeddedBroker.INSTANCE ==null){
throw new IllegalStateException("Not Initialized");
}
return INSTANCE;
}
/**
* Initialize singleton instance.
*
* @return
* @throws Exception
*/
public static EmbeddedBroker initialize(String brokerName) throws Exception{
if(EmbeddedBroker.INSTANCE ==null){
EmbeddedBroker.INSTANCE=new EmbeddedBroker(brokerName, false);
}
else{
throw new IllegalStateException("Already Initialized");
}
return INSTANCE;
}
/**
* Initialize singleton instance.
*
* @return
* @throws Exception
*/
public static EmbeddedBroker initialize(String brokerName, boolean enableTCPConnector) throws Exception{
if(EmbeddedBroker.INSTANCE ==null){
EmbeddedBroker.INSTANCE=new EmbeddedBroker(brokerName, enableTCPConnector);
}
else{
throw new IllegalStateException("Already Initialized");
}
return INSTANCE;
}
/**
* Private constructor
* @throws Exception
*/
private EmbeddedBroker(String brokerName, boolean enableTCPConnector) throws Exception{
//-- By default a broker always listens on vm://<broker name>
this.broker = new BrokerService();
this.broker.setBrokerName(brokerName);
//-- Enable Advisory Support. Its true by default, but this is to explicitly mention it for documentation purposes
this.broker.setAdvisorySupport(true);
/* Create non-persistent broker to use inMemory Store,
* instead of KAHA or any other persistent store.
* See Section 4.6.1 of ActiveMQInAction
*/
this.broker.setPersistent(false);
//-- 64 MB
this.broker.getSystemUsage().getMemoryUsage().setLimit(64*1024*1024);
//-- Set the Destination policies
PolicyEntry policy = new PolicyEntry();
//-- Set a memory limit of 4mb for each destination
policy.setMemoryLimit(4 * 1024 *1024);
//-- Disable flow control
policy.setProducerFlowControl(false);
PolicyMap pMap = new PolicyMap();
//-- Configure the policy
pMap.setDefaultEntry(policy);
this.broker.setDestinationPolicy(pMap);
if(enableTCPConnector)
broker.addConnector("tcp://localhost:61616");
//-- Start the Broker.
this.broker.start();
}
}
public class NonPersistentProducer {
private final MessageProducer producer;
private final Session session;
public NonPersistentProducer(String brokerUrl, String topic) throws JMSException{
//-- Tell the connection factory to connect to a broker and topic passed in.
ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory(brokerUrl);
//-- Disable message copying
cf.setCopyMessageOnSend(false);
Connection connection = cf.createConnection();
connection.start();
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Topic topicObj = session.createTopic(topic);
producer = session.createProducer(topicObj);
//-- Send non-persistent messages
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
}//-- Producer
/**
* @return the producer
*/
public MessageProducer getProducer() {
return producer;
}
/**
* @return the session
*/
public Session getSession() {
return session;
}
}