Я пробую пример начала работы с Java на https://docs.microsoft.com/en-us/azure/iot-hub/iot-hub-java-java-getstarted в концентраторе Интернета вещей с 4 разделами.
Я адаптировал часть о получении событий, создав 1 экземпляр клиента и вызвав client.createReceiver для каждого раздела, как показано ниже:
public class App {
private static String connStr = "...";
public static void main(String[] args) throws IOException, EventHubException, ExecutionException, InterruptedException {
EventHubClient client = createClient();
String[] partitionIds = client.getRuntimeInformation()
.thenApply(eventHubRuntimeInformation -> eventHubRuntimeInformation.getPartitionIds())
.get();
client.createReceiver(DEFAULT_CONSUMER_GROUP_NAME, "0", Instant.now())
.thenAccept(receiverHandler("0"));
client.createReceiver(DEFAULT_CONSUMER_GROUP_NAME, "1", Instant.now())
.thenAccept(receiverHandler("0"));
client.createReceiver(DEFAULT_CONSUMER_GROUP_NAME, "2", Instant.now())
.thenAccept(receiverHandler("2"));
client.createReceiver(DEFAULT_CONSUMER_GROUP_NAME, "3", Instant.now())
.thenAccept(receiverHandler("3"));
System.out.println("Press ENTER to exit.");
System.in.read();
try {
client.closeSync();
System.exit(0);
} catch (Exception e) {
System.exit(1);
}
}
private static Consumer<PartitionReceiver> receiverHandler(String partitionId) {
return receiver -> {
System.out.println("** Created receiver on partition " + partitionId);
try {
while (true) {
receiver.receive(10)
.thenAccept(receivedEvents -> {
int batchSize = 0;
if (receivedEvents != null) {
System.out.println("Got some evenst");
for (EventData receivedEvent : receivedEvents) {
System.out.println(String.format("Offset: %s, SeqNo: %s, EnqueueTime: %s",
receivedEvent.getSystemProperties().getOffset(),
receivedEvent.getSystemProperties().getSequenceNumber(),
receivedEvent.getSystemProperties().getEnqueuedTime()));
System.out.println(String.format("| Device ID: %s",
receivedEvent.getSystemProperties().get("iothub-connection-device-id")));
System.out.println(String.format("| Message Payload: %s",
new String(receivedEvent.getBytes(), Charset.defaultCharset())));
batchSize++;
}
}
System.out.println(String.format("Partition: %s, ReceivedBatch Size: %s", partitionId, batchSize));
}
).get();
}
} catch (Exception e) {
System.out.println("Failed to receive messages: " + e.getMessage());
}
};
}
private static EventHubClient createClient() {
EventHubClient client = null;
try {
client = EventHubClient.createFromConnectionStringSync(connStr);
} catch (Exception e) {
System.out.println("Failed to create client: " + e.getMessage());
System.exit(1);
}
return client;
}
}
События, отправляемые с имитируемого устройства, поступают в раздел 3. Проблема заключается в том, что события не принимаются в следующих ситуациях:
- устройство не отправляет события при запуске приемника событий
- устройство отправляет события при запуске приемника событий, но затем перезапускается. В этом случае приведенный выше код перестает получать события после перезагрузки устройства.
Вышеупомянутые проблемы не возникают, когда мы подключаемся только к разделу 3.
Вышеуказанные проблемы не возникают, когда мы подключаемся только к разделу 3 и 1 другому разделу.
Вышеуказанные проблемы возникают при подключении к разделу. 3 и 2 или 3 другие перегородки.
Любые подсказки?
