Мы включили функцию диагностики в нашей учетной записи пакета для потоковой передачи событий в концентратор событий, который мы собираем в нашем приложении для выполнения действий, основанных на состояниях пакетных задач. Однако мы замечаем, что соединение закрывается автоматически (возможно, из-за того, что за ночь не происходит никаких событий), и, следовательно, нам приходится периодически возвращать сервер обратно, чтобы снова получать события / сообщения.
Мы по-прежнему полагаемся на Java 7, и вот зависимости, которые мы добавили для пакетной обработки:
//azure dependency
compile('com.microsoft.azure:azure-storage:7.0.0')
compile('com.microsoft.azure:azure-batch:5.0.1') {
//do not get transitive dependency com.nimbusds:nimbus-jose-jw because spring security still rely on old version of it
excludes group: 'com.nimbusds', module: 'nimbus-jose-jw'
}
compile('com.fasterxml.jackson.core:jackson-core:2.9.8')
compile('org.apache.qpid:qpid-amqp-1-0-common:0.32')
compile('org.apache.qpid:qpid-amqp-1-0-client:0.32')
compile('org.apache.qpid:qpid-amqp-1-0-client-jms:0.32')
compile('org.apache.qpid:qpid-jms-client:0.40.0')
compile('org.apache.geronimo.specs:geronimo-jms_1.1_spec:1.1.1')
//end of azure dependency
А вот фрагмент кода, который устанавливает соединение, на самом деле мы использовали приведенный здесь пример кода: http://theitjourney.blogspot.com/2015/12/sendreceive-messages-using-amqp-in-java.html, так как мы не смогли найти никакого рабочего примера для Java 7 в самом документе Azure.
/**
* Set up connection to the service bus using AMQP mechanism.
* NOTE: Messages received from the message bus are not guaranteed to follow order.
* */
MessageConsumer initiateConsumer(MessageListener messageListener, Integer partitionInx, BatchEventHubConfig batchEventHubConfig) {
// set up JNDI context
String queueName = "EventHub"
String connectionFactoryName = "SBCFR"
Hashtable<String, String> hashtable = new Hashtable<>()
hashtable.put("connectionfactory.${connectionFactoryName}", batchEventHubConfig.getAMQPConnectionURI())
hashtable.put("queue.${queueName}", "${batchEventHubConfig.name}/ConsumerGroups/${batchEventHubConfig.consumerGroup}/Partitions/${partitionInx}")
hashtable.put(Context.INITIAL_CONTEXT_FACTORY, "org.apache.qpid.amqp_1_0.jms.jndi.PropertiesFileInitialContextFactory")
Context context = new InitialContext(hashtable)
ConnectionFactory factory = (ConnectionFactory) context.lookup(connectionFactoryName)
Destination queue = (Destination) context.lookup(queueName)
Connection connection = factory.createConnection(batchEventHubConfig.sasPolicyName, batchEventHubConfig.sasPolicyKey)
connection.setExceptionListener(new BatchExceptionListener())
connection.start()
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE)
MessageConsumer messageConsumer = session.createConsumer(queue)
messageConsumer.setMessageListener(messageListener)
messageConsumer
}
Так есть ли способ отследить, было ли соединение закрыто, и если да, перезапустить соединение снова?
Также будет полезна любая информация для дальнейшей диагностики этой проблемы.