我有一个producer consumer使用WildflyAS的设置JMS，生产者newFixedThreadPool(126)每 1 分钟使用一次，每个线程从REST服务中提取数据并将其推送到ASHornetQ上。Wildfly
然后在消费者方面，我有一个消费者类，它消耗HornetQ和一个简单的Parser类中的消息，对于数据库插入，我Parser在我的实例中实例化一个类型的对象，onMessage()然后将消息传递给它，消息在JSON并且我的解析器类循环通过它获取值并将它们插入到我的数据库中。
消费者：
public void Consume(Consumer asyncReceiver) throws Throwable {
try {
/** Get the initial context */
final Properties props = new Properties();
/** If debugging in IDE the properties are acceded this way */
if(debug){
InputStream f = getClass().getClassLoader().getResourceAsStream("consumer.properties");
props.load(f);
}
/** If running the .jar artifact the properties are acceded this way*/
else{
File jarPath = new File(getClass().getProtectionDomain().getCodeSource().getLocation().getPath());
String propertiesPath = jarPath.getParentFile().getAbsolutePath();
props.load(new FileInputStream(propertiesPath + File.separator + "consumer.properties"));
}
/** These few lines should be removed and setup in the properties file*/
props.put(Context.INITIAL_CONTEXT_FACTORY, props.getProperty("INITIAL_CONTEXT_FACTORY"));
props.put(Context.PROVIDER_URL, props.getProperty("PROVIDER_URL"));
props.put(Context.SECURITY_PRINCIPAL, props.getProperty("DEFAULT_USERNAME"));
props.put(Context.SECURITY_CREDENTIALS, props.getProperty("DEFAULT_PASSWORD"));
context = new InitialContext(props);
/** Lookup the queue object */
Queue queue = (Queue) context.lookup(props.getProperty("DEFAULT_DESTINATION"));
/** Lookup the queue connection factory */
ConnectionFactory connFactory = (ConnectionFactory) context.lookup(props.getProperty("DEFAULT_CONNECTION_FACTORY"));
/** Create a queue connection */
connection = connFactory.createConnection(props.getProperty("DEFAULT_USERNAME"), props.getProperty("DEFAULT_PASSWORD"));
/** Create a queue session */
queueSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
/** Create a queue consumer */
msgConsumer = queueSession.createConsumer(queue);
/** Set an asynchronous message listener */
msgConsumer.setMessageListener(asyncReceiver);
/** Set an asynchronous exception listener on the connection */
connection.setExceptionListener(asyncReceiver);
/** Start connection */
connection.start();
/** Wait for messages */
System.out.println("waiting for messages");
for (int i = 0; i < 47483647; i++) {
Thread.sleep(1000);
System.out.print(".");
}
System.out.println();
} catch (Exception e) {
log.severe(e.getMessage());
throw e;
}finally {
if (context != null) {
context.close();
}
if (queueSession != null)
{ queueSession.close();
}
if(msgConsumer != null){
msgConsumer.close();
}
if (connection != null) {
connection.close();
}
}
}
@Override
public void onMessage(Message message) {
TextMessage msg = (TextMessage) message;
try {
Parser parser = new Parser();
parser.parseApplication(msg.getText());
} catch (Exception e) {
e.printStackTrace();
}
}
解析器：
public void parseApplication(String NRData) throws Exception {
DBConnection db = DBConnection.createApplication();
db.getConnection();
JsonFactory factory = new JsonFactory();
ObjectMapper mapper = new ObjectMapper(factory);
JsonNode rootNode = mapper.readTree(NRData);
Iterator<Map.Entry<String, JsonNode>> fieldsIterator = rootNode.fields();
while (fieldsIterator.hasNext()) {
Map.Entry<String, JsonNode> field = fieldsIterator.next();
String envName = field.getKey();
JsonNode appValue = field.getValue();
JSONArray jsonArray = new JSONArray(appValue.toString());
String appName = jsonArray.getString(0);
String appID = jsonArray.getString(1);
JSONObject json = jsonArray.getJSONObject(2);
JSONObject metricsData = json.getJSONObject("metric_data");
JSONArray metrics = metricsData.getJSONArray("metrics");
JSONObject array1 = metrics.getJSONObject(0);
JSONArray timeslices = array1.getJSONArray("timeslices");
for (int i = 0; i < timeslices.length(); i++) {
JSONObject array2 = timeslices.getJSONObject(i);
JSONObject values = array2.getJSONObject("values");
// Instant from = array2.getString("from");
Instant from = TimestampUtils.parseTimestamp(array2.get("from").toString(), null);
Instant to = TimestampUtils.parseTimestamp(array2.get("to").toString(), null);
Iterator<String> nameItr = values.keys();
while (nameItr.hasNext()) {
String name = nameItr.next();
System.out.println(
"\nEnv name: " + envName +
"\nApp name: " + appName +
"\nApp ID: " + appID +
"\nRequest per minute: " + values.getDouble(name) +
"\nFrom: " + from + " To: " + to);
ThroughputEntry TP = new ThroughputEntry();
TP.setThroughput(values.getDouble(name));
TP.setEnvironment(envName);
TP.setName(appName);
TP.setRetrieved(from);
TP.setPeriodEnd(to);
db.addHistory(TP);
}
}
}
}
数据库连接：
public class DBConnection {
private final String table;
/**
* Set the table name for applications
*/
public static DBConnection createApplication() {
return new DBConnection("APPLICATIONDATA");
}
public DBConnection(String table) {
this.table = String.format("NRDATA.%s", table);
}
public Connection getConnection() throws IllegalAccessException,
InstantiationException, ClassNotFoundException, SQLException {
try {
Class.forName("COM.ibm.db2os390.sqlj.jdbc.DB2SQLJDriver");
} catch (ClassNotFoundException e) {
e.printStackTrace();
}
System.out.println("Connecting to database...");
Connection connection = DriverManager.getConnection(DB2url, user, password);
System.out.println("From DAO, connection obtained ");
return connection;
}
public boolean addHistory(ThroughputEntry entry) throws Exception {
try (Connection connection = getConnection()) {
Statement statement = connection.createStatement();
return 0 < statement
.executeUpdate(String
.format("INSERT INTO " + table
+ "(ID, RETRIEVED, PERIOD_END, ENVIRONMENT, APPNAME, THROUGHPUT)"
+ "VALUES ('%s', '%s', '%s', '%s', '%s', %s)",
entry.getUUID(), entry.getRetrieved(),
entry.getPeriodEnd(), entry.getEnvironment(),
entry.getName(), entry.getThroughput()));
}
}
}
所以我的问题是我在WildflyAS 上出现内存泄漏，我认为问题可能出在我的消费者身上。
所以有几个问题：
onMessage()在将消息插入数据库之前，我应该在消费者的方法中缓冲接收到的消息吗？
如果我收到太多消息，这会导致泄漏吗？Ack消费者会向WildflyAS发送任何类型的 og吗？
我让消费者无限期地循环运行，也许这是错误的，也许它应该休眠或等待。
我已经 2 天试图解决这个问题，任何帮助将不胜感激。