Медленная доставка сообщений в приложении Java с использованием oracle Advanced Queue ⇐ JAVA

[Anonymous]

У меня есть приложение Java, которое подключается к Oracle Advanced Queue (AQ) и прослушивает различные типы событий. Сервер Java и сервер Oracle находятся на разных машинах, но в одной и той же сети обработки данных. Despite a very low event generation rate, we frequently observe delays of around 10 seconds between enqueueing an event and receiving it in Java.
To measure this, we:

[*]Insert a log entry with a timestamp into a table just before pushing the event onto the AQ queue.
Log another timestamp in our Java MessageListener, как только появится сообщение. (См. Используемые зависимости ниже) (См. Реализацию подключения ниже)

Код: Выделить всё

com.oracle.database.messaging
aqapi-jakarta
23.3.1.0


jakarta.jms
jakarta.jms-api
3.1.0


jakarta.transaction
jakarta.transaction-api
2.0.1

фабрика соединения

Код: Выделить всё

@Component
public class QueueConn {
private static final Logger LOGGER = LoggerFactory.getLogger(QueueConn.class);

@Override
public QueueConnectionWrapper createQueueConnectionWrapper()
throws JMSException, SQLException {
Connection dbConn = getConnection();
try {
QueueConnection aqConn =
AQjmsQueueConnectionFactory.createQueueConnection(dbConn);
return new QueueConnectionWrapper(dbConn, aqConn);
} catch (JMSException | RuntimeException e) {
close(dbConn);
if (e instanceof JMSException) throw (JMSException) e;
throw new SQLException("Failed to create AQ queue connection", e);
}
}

@Override
public void close(Connection c) {
if (c != null) {
try {
c.close();
} catch (SQLException e) {
LOGGER.error("Failed to close DB connection", e);
}
}
}

@Override
public QueueSession createQueueSession(QueueConnection queueConnection)
throws JMSException {
return queueConnection.createQueueSession(
false, Session.CLIENT_ACKNOWLEDGE);
}

@Override
public MessageConsumer createMessageConsumer(
String queueName,
QueueSession queueSession,
String schemaName
) throws JMSException, SQLException {
Queue queue = ((AQjmsSession) queueSession)
.getQueue(schemaName, queueName);
return queueSession.createConsumer(queue);
}

private Connection getConnection() throws SQLException {
DatabaseConfig cfg = /* load config */;
PoolDataSource pds = PoolDataSourceFactory.getPoolDataSource();
pds.setConnectionFactoryClassName("oracle.jdbc.pool.OracleDataSource");
pds.setURL(cfg.connectionString());
pds.setUser(cfg.username());
pds.setPassword(cfg.password());
pds.setValidateConnectionOnBorrow(cfg.validateConnectionOnBorrow());
pds.setInitialPoolSize(cfg.initialPoolSize());
pds.setMaxPoolSize(cfg.maxPoolSize());
pds.setMaxIdleTime(cfg.maxIdleTime());
pds.setSecondsToTrustIdleConnection(cfg.secondsToTrustIdleConnection());
return pds.getConnection();
}
}

реализация прослушивателя

Код: Выделить всё

public class Listener implements MessageListener {
private static final Logger LOGGER = LoggerFactory.getLogger(Listener.class);

@Override
public void onMessage(Message message) {
// log timestamp and process message
}
}
< /code>
Очередь в Oracle была создана с использованием < /p>
begin
dbms_aqadm.create_queue_table (
Queue_table => 'my_event_jms_map_table',
Queue_payload_type => 'SYS.AQ$_JMS_MAP_MESSAGE');
end;
/

begin
dbms_aqadm.create_queue(
Queue_name => 'my_event_jms_map_queue',
Queue_table => 'my_event_jms_map_table');
end;
< /code>
У нас нет другой специальной конфигурации вокруг очереди.   Если есть определенные параметры, мы должны проверить, пожалуйста, сообщите нам об этом, так как это может быть хорошей подсказкой.  Мы не зацикливаемся и не избираем, мы используем библиотеку Oracle Java, которую они предоставляют для подключения к расширенным очереди и реагируем на обработчик onmessage 

, как показано выше.


Подробнее здесь: https://stackoverflow.com/questions/796 ... nced-queue