我正在尝试找出以下错误消息的解决方案，而不将默认连接大小从 1000 增加到 2000 或更多。
最近，当大约 1000 条消息发送到代理时，我遇到了以下错误，延迟了 5 分钟，如下面的代码所示。
WARN | Could not accept connection : Exceeded the maximum number of allowed client connections. See the 'maximumConnections' property on the TCP transport configuration URI in the ActiveMQ configuration file (e.g., activemq.xml) | org.apache.activemq.broker.TransportConnector | ActiveMQ Transport Server Thread Handler: tcp://0.0.0.0:61616?maximumConnections=1000&wireFormat.maxFrameSize=104857600
以下是持续监听ActiveMQ的代码，一旦看到
COMPLETEelse在 else 块内，我想在发送完消息后关闭连接来进行测试。所以我已经关闭了finally块内的连接，如下所示。这是正确的做法吗？
@Component
public class DownloadConsumer {
@Autowired
private JavaMailSender javaMailSender;
// one instance, reuse
private final CloseableHttpClient httpClient = HttpClients.createDefault();
Connection connection;
// Working Code with JMS 2.0
@JmsListener(destination = "MessageProducerJMSV1")
public void processBrokerQueues(String message) throws DaoException, JMSException {
try {
RequestDao requestDao = (RequestDao) context.getBean("requestDao");
String receivedStatus = requestDao.getRequestStatus(message);
//Retrieve Username from the message to include in an email
String[] parts = message.split("#");
String userName = parts[1].trim();
//Retrieve personnelID from the message to include in the webservice calls
String personnelID = parts[3].trim();
//Before sending this message, do the check for COMPLETE or ERROR etc
if(receivedStatus.equals("COMPLETE")) {
String latestUUID = requestDao.getUUID();
logger.info("Received UUID in Controller is as follows! ");
logger.info(latestUUID);
requestDao.sendMessage(message,latestUUID);
logger.info("Received status is COMPLETE! ");
logger.info("Sending email to the user! ");
String emailMessage = "Dear "+userName+",<p>Your files are ready. </p><p> Thanks,<br/> Jack/p>";
String recipientEmail = userName+"@organization.com";
/*****************************************************\
// START: EMAIL Related Code
*******************************************************/
MimeMessage msg = javaMailSender.createMimeMessage();
MimeMessageHelper helper = new MimeMessageHelper(msg, true);
helper.setFrom("[email protected]");
helper.setTo(recipientEmail);
helper.setSubject("Requested Files !");
helper.setText(emailMessage,true);
javaMailSender.send(msg);
}
else {
// Getting JMS connection from the server and starting it
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
connection = connectionFactory.createConnection();
connection.start();
Session session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
// Destination represents here our queue 'MessageProducerJMSV1' on the JMS server
Destination destination = session.createQueue(subject);
MessageProducer producer = session.createProducer(destination);
//Sending message to the queue
TextMessage toSendMessage = session.createTextMessage(message);
long delay = 300 * 1000;
long period =300 * 1000;
toSendMessage.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, delay);
producer.send(toSendMessage);
}
}
catch(Throwable th){
th.printStackTrace();
}
finally {
connection.close();
}
}
// URL of the JMS server. DEFAULT_BROKER_URL will just mean that JMS server is on localhost
private static String url = ActiveMQConnection.DEFAULT_BROKER_URL;
private static String subject = "MessageProducerJMSV1"; //Queue Name
// default broker URL is : tcp://localhost:61616"
private static ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("ApplicationContext.xml");
private static final Logger logger = LoggerFactory.getLogger(DownloadConsumer.class);
}
您收到“超出允许的最大客户端连接数”的原因是因为您正在创建连接而不是关闭它们。换句话说，您的应用程序正在“泄漏”连接。为了修复泄漏，您需要关闭连接。在
finallynullconnectionfinally {
if (connection != null) {
connection.close();
}
}
也就是说，值得注意的是，创建和关闭 JMS 连接和会话以及生成器以发送“单个”消息是一种众所周知的反模式。如果您缓存连接（例如在 static 变量中）并重新使用它，那就更好了。例如：
@Component
public class DownloadConsumer {
@Autowired
private JavaMailSender javaMailSender;
// one instance, reuse
private final CloseableHttpClient httpClient = HttpClients.createDefault();
private static Connection connection;
private static Object connectionLock = new Object();
// URL of the JMS server. DEFAULT_BROKER_URL will just mean that JMS server is on localhost
private static String url = ActiveMQConnection.DEFAULT_BROKER_URL;
private static String subject = "MessageProducerJMSV1"; //Queue Name
// default broker URL is : tcp://localhost:61616"
private static ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("ApplicationContext.xml");
private static final Logger logger = LoggerFactory.getLogger(DownloadConsumer.class);
// Working Code with JMS 2.0
@JmsListener(destination = "MessageProducerJMSV1")
public void processBrokerQueues(String message) throws DaoException, JMSException {
try {
RequestDao requestDao = (RequestDao) context.getBean("requestDao");
String receivedStatus = requestDao.getRequestStatus(message);
//Retrieve Username from the message to include in an email
String[] parts = message.split("#");
String userName = parts[1].trim();
//Retrieve personnelID from the message to include in the webservice calls
String personnelID = parts[3].trim();
//Before sending this message, do the check for COMPLETE or ERROR etc
if (receivedStatus.equals("COMPLETE")) {
String latestUUID = requestDao.getUUID();
logger.info("Received UUID in Controller is as follows! ");
logger.info(latestUUID);
requestDao.sendMessage(message, latestUUID);
logger.info("Received status is COMPLETE! ");
logger.info("Sending email to the user! ");
String emailMessage = "Dear " + userName + ",<p>Your files are ready. </p><p> Thanks,<br/> Jack/p>";
String recipientEmail = userName + "@organization.com";
/*****************************************************\
// START: EMAIL Related Code
*******************************************************/
MimeMessage msg = javaMailSender.createMimeMessage();
MimeMessageHelper helper = new MimeMessageHelper(msg, true);
helper.setFrom("[email protected]");
helper.setTo(recipientEmail);
helper.setSubject("Requested Files !");
helper.setText(emailMessage, true);
javaMailSender.send(msg);
} else {
try {
createConnection();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// Destination represents here our queue 'MessageProducerJMSV1' on the JMS server
Destination destination = session.createQueue(subject);
MessageProducer producer = session.createProducer(destination);
//Sending message to the queue
TextMessage toSendMessage = session.createTextMessage(message);
long delay = 300 * 1000;
long period = 300 * 1000;
toSendMessage.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, delay);
producer.send(toSendMessage);
} catch (Throwable th) {
th.printStackTrace();
synchronized (connectionLock) {
// if there are any problems close the connection and it will be re-created next time
if (connection != null) {
connection.close();
connection = null;
}
}
}
}
} catch (Throwable th) {
th.printStackTrace();
}
}
private void createConnection() {
synchronized (connectionLock) {
if (connection == null) {
// Getting JMS connection from the server
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
connection = connectionFactory.createConnection();
}
}
}
}
您会注意到，在此代码中没有
finally块来关闭连接。这是故意的，因为此代码的重点是“保持连接打开”，这样它就不会打开和关闭连接来发送单个消息。连接在调用之间被重复使用。连接关闭的唯一时间是捕获
Throwable发送消息，则没有理由在 JMS 连接上调用
start()。
start()问题是下面的代码，它将打开每条消息的新连接 - 理想情况下，您应该调用一次（如果连接过期则再次调用）。
从中获取会话并关闭会话。根据使用情况，您甚至可以保持会话更长时间。