我的问题是如果 jms 队列(Wildfly 10)被大量填充,我该如何处理服务器上的负载。问题是消息不断被添加到JMS队列中,但是它们是被一个一个处理的。写的逻辑是只一个一个处理消息。我们不能一次处理多个消息,逻辑不能改变。但这会导致服务器负载增加,因为队列加载了许多消息。此时我该如何管理队列。
这是我用于生产和消费的示例代码:-
以下代码每 100 毫秒运行一次,并从数据库中获取 100 条消息并发送到队列。
Arraylist arrMessages=GetMessagesFromdatabase();
for(MessageObject obj:arrMessages) sendMessagetoQueue(obj);
生产者代码:-
void sendMessagetoQueue(MessageObject messageObject){
ConnectionFactory connectionFactory = JMSConstants.getConnFactory.getJmsConnectionFactory();
Context initialContext = JMSConstants.getConnFactory.getInitialcontext();
Connection connection = null;
String destinationName = "java:/jms/queue/Queue";
MessageProducer publisher = null;
Session session = null;
try {
connection = connectionFactory.createConnection();
Queue queue = (Queue) initialContext
.lookup(destinationName);
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
publisher = session.createProducer(queue);
connection.start();
ObjectMessage objMessage = session.createObjectMessage(messageObject);
publisher.send(objMessage);
} catch (Exception exc) {
exc.printStackTrace();
LOG.error(exc);
}
}
然后另一个计划每 100 毫秒运行一次,以消耗队列中的消息并将其转发以逐个处理。
消费者守则:-
ConnectionFactory connectionFactory = JMSConstants.getConnFactory.getJmsConnectionFactory();
Context initialContext = JMSConstants.getConnFactory.getInitialcontext();
Connection connection = null;
String destinationName = "java:/jms/queue/Queue";
Session session = null;
MessageConsumer consumer = null;
Boolean checkFlag = true;
QueueBrowser queueBrowserconnect = null;
try {
connection = connectionFactory.createConnection();
Queue queue = (Queue) initialContext.lookup(destinationName);
session = connection.createSession();
consumer = session.createConsumer(queue);
Queue senderqueueconnect = (Queue) initialContext.lookup(senderQueueconnect);
queueBrowserconnect = session.createBrowser(senderqueueconnect);
connection.start();
ObjectMessage objectMessage = (ObjectMessage) consumer.receive(1);
if (objectMessage != null) {
objectMessage.acknowledge();
MessageObject messageobject= (MessageObject) objectMessage.getObject();
//Send object for processing
messageService.processInputFromQueue(messageobject);
}
} catch (Exception exc) {
exc.printStackTrace();
LOG.error(exc.getMessage());
}