野蝇10.1.0的内存泄漏(java.lang.ref.Finalizer / ActiveMQConnection)

Question

我们有一个java应用程序，它发送(JobsController.java)并通过JMS接收消息(JMSMessageListener.java)。在常量加载下运行应用程序24小时并接受堆转储后，我观察到内存使用量不断增加，而应用程序在空闲状态时不会松手。我知道这将导致内存不足的java堆问题。

JobsController是一个ejb无状态bean，它的资源在每次调用后都会被正确地销毁。JMSMessageListener由ejb池处理，并且它的实例被重用。

我从java堆转储中可以看到的嫌疑人是

1. EJB注入导致内存泄漏，https://blog.akquinet.de/2017/01/04/dont-get-trapped-into-a-memory-leak-using-cdi-instance-injection/。
2. ActiveMQConnection.finalize()。如果是罪魁祸首，那么所有这些野蝇活动部署都必须发生。任何暗示都是感激的。

ActiveMQConnection.java

@Override  
protected final void finalize() throws Throwable {  
   if (!closed) {  
       if (this.factoryReference.isFinalizeChecks()) {  
           ActiveMQJMSClientLogger.LOGGER.connectionLeftOpen(creationStack);  
       }  
       close();  
}

​

​

JobsController

@无国籍

公共类JobsController {

@Inject  
private JMSContext jmsContext;  
private Connection connection;  
private Session session;  
private MessageProducer jmsProducer;  

@Resource(lookup = "java:/ConnectionFactory")  
private ConnectionFactory connectionFactory;  

@Resource(lookup = JAVA_JMS_JOB_QUEUE)  
private Queue jobQueue;  

@Resource(lookup = JAVA_JMS_QUEUE)  
private Queue progressQueue;  

@PreDestroy  
void release() {  
    try {  
        if (jmsProducer != null) {  
            jmsProducer.close();  
        }  
        if (session != null) {  
            session.close();  
        }  
        if (jmsContext != null) {  
            jmsContext.close();  
        }  
        if (connection !=null) {  
            connection.close();  
        }  
    } catch (JMSException e) {  
        LOG.warn("failed to close JMS resources: {}", e.getMessage());  
    }  
}  

public synchronized MessageProducer getJmsProducer() {  
    if (jmsProducer == null) {  
        try {  
            connection = connectionFactory.createConnection();  
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);  
            jmsProducer = session.createProducer(jobQueue);  


            connection.start();  
        } catch (JMSException e) {  
            LOG.error("failed to setup JMS message producer: {}", e.getMessage());  
        }  
    }  
    return jmsProducer;  
}
public void addMessageToProgressQueue(ProgressMessage progressMessage) {
    ObjectMessage objectMessage = jmsContext.createObjectMessage(progressMessage);
    try {
        getJmsProducer().send(progressQueue, objectMessage);
    } catch (JMSException e) {
        LOG.error("failed to send progress message {}: {}", objectMessage, e.getMessage());
    }
}  

}

JMSMessageListener.java

@MessageDriven(name = "JMSMessageListener", mappedName = JAVA_JMS_QUEUE, activationConfig = {  
        @ActivationConfigProperty(  
                propertyName = "acknowledgeMode",  
                propertyValue = "Auto-acknowledge"),  
        @ActivationConfigProperty(  
                propertyName = "destinationType",  
                propertyValue = "javax.jms.Queue"),  
        @ActivationConfigProperty(  
                propertyName = "destination",  
                propertyValue = JAVA_JMS_QUEUE)  


})  
public class JMSMessageListener implements MessageListener {  

    private static Logger LOG = LoggerFactory.getLogger(JMSMessageListener.class);  

    @EJB  
    private JobsController jobsController;  

    private final ObjectMapper progressMessageMapper;  

    public JMSMessageListener() {  
        progressMessageMapper = new ObjectMapper();  
        progressMessageMapper.configure(JsonParser.Feature.ALLOW_SINGLE_QUOTES, true);  
    }  

    @Override  
    public void onMessage(Message message) {  
        ProgressMessage progressMessage = null;  
        try {  
            if (message instanceof BytesMessage) {  
                BytesMessage bytesMessage = (BytesMessage) message;  
                int TEXT_LENGTH = new Long(bytesMessage.getBodyLength()).intValue();  
                byte[] textBytes = new byte[TEXT_LENGTH];  
                bytesMessage.readBytes(textBytes, TEXT_LENGTH);  


                String progressText = new String(textBytes, "UTF-8");  

                progressText = progressText.replaceAll("'totalSteps': None", "'totalSteps': 0");  
                progressMessage = progressMessageMapper.readValue(progressText, ProgressMessage.class);  
            } else if (message instanceof ObjectMessage) {  
                progressMessage = message.getBody(ProgressMessage.class);  
            }  
            if (progressMessage != null) {  

                jobsController.sendProgressMessage(progressMessage);  
            } else {  
                LOG.error("An empty progress message was received");  
            }  
        } catch (JMSException | IOException e) {  
            LOG.error("failed to process progress message: {}", e.getMessage(), e);  
        }  
    }  
}

Answer 1

几件事：

• 您正在注入一个JMSContext，但从未使用它(至少在您粘贴的代码中如此)。这似乎是个错误。
• 如果您不打算使用注入的JMSContext，而是使用注入的ConnectionFactory，那么您应该注入"java:/JmsXA“而不是"java:/ConnectionFactory”，因为它是一个池连接工厂>。为"java:/ConnectionFactory“发送的每条消息创建一个连接是一种反模式，因为它没有池化。此外，我假设您希望使用XA事务，以便MDB中的消息消耗和发送是原子性的，并且不适用于"java:/ConnectionFactory“。