Блог

Нашему процессу, использующему ActiveMQ, не хватает памяти, и он умирает.

У нас есть тема ActiveMQ с одним отправителем и двумя получателями. На первый взгляд, все работает нормально — сообщения отправляются и принимаются обоими получателями, но в конечном итоге мы исчерпываем всю память. Дамп кучи показывает 1,362 миллиона экземпляров, каждый из LinkedList$Node, AtomicReference, ActiveMQObjectMessage, MessageId и MessageDispatch. Между тем очередь сообщений на стороне клиента пуста или почти пуста повсюду. Я думаю, что 1,362M может быть в списке, который отслеживает неподтвержденные сообщения, ожидающие подтверждения. В теме указано значение AUTO_ACKNOWLEDGE, поэтому мы пытаемся подтвердить, но, возможно, сбой. (jmsSession = jmsConnection.CreateSession(false, сессия.AUTO_ACKNOWLEDGE);)

Дамп кучи показывает мусор, который, по-видимому, связан с буферизацией входящих сообщений клиента, существует в скромных количествах (несколько тысяч). Кажется, это согласуется с количеством этих объектов, которые накапливаются в игрушечной программе, которую мы настроили для отправки и использования похожих объектов. Они накапливаются некоторое время, затем получают GC’d, и память никогда не увеличивается значительно в игрушке или в сбойной программе.

Правильно ли предположить, что пять типов объектов связаны с ACKS, и если да, то что может привести к тому, что объекты останутся в этой структуре, несмотря на то, что они, по-видимому, полностью используются обоими потребителями? Есть ли какой-то способ, которым мы могли бы отменить AUTO_ACKNOWLEDGE, который, как мы думаем, мы установили? Кстати, один потребитель является синхронным, используя receive(), а другой — асинхронным, используя onMessage() .

Одним из возможных вводящих в заблуждение симптомов является то, что графический интерфейс ActiveMQ показывает, что объекты удаляются из очереди только один раз, несмотря на присутствие двух потребителей. Игрушка показывает два удаления из очереди для каждой очереди. Однако сама программа говорит, что они считываются ожидаемое количество раз.

 // creating the async consumer.

connAmq = createActiveMqConnection();
connAmq.start();
session = connAmq.createSession(true, Session.AUTO_ACKNOWLEDGE);
Destination topic =        session.createTopic(appProperties.getActiveMqTopicQuotesName());
MessageConsumer consumer = session.createConsumer(topic); 
consumer.setMessageListener(this);

public void onMessage(Message message) {
    ...     
    try {   
            if (message instanceof ObjectMessage) {
                    ObjectMessage msg = (ObjectMessage)message;
                    if (msg instanceof Foo) {
                            Foo quote = (Foo)msg.getObject();
                            ...
                    }       
            }
    }
    ...
}

// creating the sync consumer

jmsConnection = mActiveMQConnectionFactory.createTopicConnection();

jmsConnection.start();
jmsSession = jmsConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
jmsDestination = jmsSession.createTopic(name);
jmsMessageConsumer = jmsSession.createConsumer(jmsDestination);


//the code for consuming looks like this for the synchronous consumer
while(true)
ObjectMessage m = (ObjectMessage) jmsMessageConsumer.receive();
if (m != null)
    Process(m.getObject());
}
  

По крайней мере, в приведенном фрагменте кода вы создали сеанс, который выполняется для асинхронного потребителя, но я не вижу вызова commit в сеансе. Биты транзакции остаются в памяти брокера, и вы в конечном итоге исчерпаете память брокера.