0

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

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

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

  1. EJB bean 注入导致内存泄漏 https://blog.akquinet.de/2017/01/04/dont-get-trapped-into-a-memory-leak-using-cdi-instance-injection/
  2. ActiveMQConnection.finalize()。如果它是罪魁祸首，那么它必须发生在所有这些 wildfly activemq 部署中。任何提示表示赞赏。

ActiveMQConnection.java

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

在此处输入图像描述 在此处输入图像描述 在此处输入图像描述

工作控制器

@Stateless 公共类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);  
        }  
    }  
}
4

1 回答 1

1

几件事：

  • 您正在注入 JMSContext 但从未使用它（至少在您粘贴的代码中）。这似乎是一个错误。
  • 如果您不打算使用注入的 JMSContext 而是使用注入的 ConnectionFactory，那么您应该注入“java:/JmsXA”而不是“java:/ConnectionFactory”，因为它是一个&tl;pooled-connection-factory>。为使用“java:/ConnectionFactory”发送的每条消息创建一个连接是一种反模式，因为它没有被池化。此外，我假设您希望使用 XA 事务，以便 MDB 中的消息消费和发送是原子的，并且不适用于“java:/ConnectionFactory”。
于 2017-09-01T20:13:16.450 回答