azure-eventhub - Java EventHubClient 在监听超过 2 个分区时不接收事件

Question

我正在一个具有 4 个分区的 iot 集线器上尝试https://docs.microsoft.com/en-us/azure/iot-hub/iot-hub-java-java-getstarted上的 java 入门示例。

client.createReceiver我通过创建 1 个客户端实例并调用每个分区来调整有关接收事件的部分，如下所示：

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。问题是在以下情况下没有收到任何事件：

1. 启动事件接收器时设备未发送事件
2. 设备在启动事件接收器时正在发送事件，但之后会重新启动。在这种情况下，上述代码在设备重启后停止接收事件

当我们只连接到分区 3
时不会出现上述问题。当我们只连接到分区 3 和 1 其他分区时不会出现上述问题。
当我们连接到分区 3 和 2 或 3 其他分区时，确实会出现上述问题。

有什么线索吗？

Answer 1

当我们连接到分区 3 和 2 或 3 其他分区时，确实会出现上述问题。

我使用您的代码失败并出现此错误“-source 1.5 java 中不支持 lambda 表达式”。

然后我编辑教程中的示例代码，让一个客户端连接到四个分区接收器（这里四个设备事件放在三个分区中，不是四个，因为设备1和设备2的事件都放在分区2中。）：

public class App 
{
    private static String connStr = "Endpoint=sb://iothub-ns-ritatestio-265731-93e2a49f65.servicebus.windows.net/;EntityPath=ritatestiothub;SharedAccessKeyName=iothubowner;SharedAccessKey=z+QM62TftPlTfwS3CnN9348X2cmMkCaEFaC1IqYpiW8=";
    private static EventHubClient client = null;

    public static void main( String[] args ) throws IOException
    {       
        try {
          client = EventHubClient.createFromConnectionStringSync(connStr);
        } catch (Exception e) {
          System.out.println("Failed to create client: " + e.getMessage());
          System.exit(1);
        }

        // Create receivers for partitions 0 and 1.
        receiveMessages("0");
        receiveMessages("1");
        receiveMessages("2");
        receiveMessages("3");
        System.out.println("Press ENTER to exit.");

        System.in.read();
        try {
          client.closeSync();
          System.exit(0);
        } catch (Exception e) {
          System.exit(1);
        }
    }

    // Create a receiver on a partition.
private static void receiveMessages(final String partitionId) {
  try {
    // Create a receiver using the
    // default Event Hubs consumer group
    // that listens for messages from now on.
    client.createReceiver(EventHubClient.DEFAULT_CONSUMER_GROUP_NAME, partitionId, Instant.now())
      .thenAccept(new Consumer<PartitionReceiver>() {
        public void accept(PartitionReceiver receiver) {
          System.out.println("** Created receiver on partition " + partitionId);
          try {
            while (true) {
              Iterable<EventData> receivedEvents = receiver.receive(100).get();

              int batchSize = 0;
              if (receivedEvents != null) {
                for (EventData receivedEvent : receivedEvents) {
                  batchSize++;
                  System.out.println(String.format("Partition: %s, ReceivedBatch Size: %s, Device ID: %s, SeqNo: %s", partitionId, batchSize,receivedEvent.getSystemProperties().get("iothub-connection-device-id"),receivedEvent.getSystemProperties().getSequenceNumber()));
                }
              }

            }
          } catch (Exception e) {
            System.out.println("Failed to receive messages: " + e.getMessage());
          }
        }
      });
    } catch (Exception e) {
      System.out.println("Failed to create receiver: " + e.getMessage());
  }

}
}

设备重启后，我可以让三个分区接收器接收事件。您可以检查以下快照：

更新：将 4 个设备更改为 1 个设备发送。尽管如此，接收方在设备重启后仍可以继续接收消息。