import com.microsoft.azure.servicebus.ServiceBusException; //导入依赖的package包/类
@Override
public void onError(Throwable throwable) {
  errorRate.inc();
  aggReadErrors.inc();

  if (throwable instanceof ServiceBusException) {
    ServiceBusException busException = (ServiceBusException) throwable;

    if (busException.getIsTransient()) {

      // Only set to transient throwable if there has been no previous errors
      eventHubHandlerError.compareAndSet(null, throwable);

      // Retry creating a receiver since error likely due to timeout
      renewPartitionReceiver(ssp);
      return;
    }
  }

  // Propagate non transient or unknown errors
  eventHubHandlerError.set(throwable);
}
 

import com.microsoft.azure.servicebus.ServiceBusException; //导入依赖的package包/类
public static void main(String[] args)
    throws ServiceBusException, IOException, ExecutionException, InterruptedException {
  Options options = new Options();
  options.addOption(
      CommandLineHelper.createOption(OPT_SHORT_EVENTHUB_NAME, OPT_LONG_EVENTHUB_NAME, OPT_ARG_EVENTHUB_NAME, true,
      OPT_DESC_EVENTHUB_NAME));

  options.addOption(
      CommandLineHelper.createOption(OPT_SHORT_NAMESPACE, OPT_LONG_NAMESPACE, OPT_ARG_NAMESPACE, true, OPT_DESC_NAMESPACE));

  options.addOption(
      CommandLineHelper.createOption(OPT_SHORT_KEY_NAME, OPT_LONG_KEY_NAME, OPT_ARG_KEY_NAME, true, OPT_DESC_KEY_NAME));

  options.addOption(
      CommandLineHelper.createOption(OPT_SHORT_TOKEN, OPT_LONG_TOKEN, OPT_ARG_TOKEN, true, OPT_DESC_TOKEN));

  CommandLineParser parser = new BasicParser();
  CommandLine cmd;
  try {
    cmd = parser.parse(options, args);
  } catch (Exception e) {
    HelpFormatter formatter = new HelpFormatter();
    formatter.printHelp(String.format("Error: %s%neh-console-consumer.sh", e.getMessage()), options);
    return;
  }

  String ehName = cmd.getOptionValue(OPT_SHORT_EVENTHUB_NAME);
  String namespace = cmd.getOptionValue(OPT_SHORT_NAMESPACE);
  String keyName = cmd.getOptionValue(OPT_SHORT_KEY_NAME);
  String token = cmd.getOptionValue(OPT_SHORT_TOKEN);

  consumeEvents(ehName, namespace, keyName, token);
}
 

import com.microsoft.azure.servicebus.ServiceBusException; //导入依赖的package包/类
private static void consumeEvents(String ehName, String namespace, String keyName, String token)
    throws ServiceBusException, IOException, ExecutionException, InterruptedException {
  ConnectionStringBuilder connStr = new ConnectionStringBuilder(namespace, ehName, keyName, token);

  EventHubClient client = EventHubClient.createFromConnectionStringSync(connStr.toString());

  EventHubRuntimeInformation runTimeInfo = client.getRuntimeInformation().get();
  int numPartitions = runTimeInfo.getPartitionCount();
  for (int partition = 0; partition < numPartitions; partition++) {
    PartitionReceiver receiver =
        client.createReceiverSync(EventHubClient.DEFAULT_CONSUMER_GROUP_NAME, String.valueOf(partition),
            PartitionReceiver.START_OF_STREAM);
    receiver.receive(10).handle((records, throwable) -> handleComplete(receiver, records, throwable));
  }
}
 

import com.microsoft.azure.servicebus.ServiceBusException; //导入依赖的package包/类
private boolean isErrorTransient(Throwable throwable) {
  if (throwable instanceof ServiceBusException) {
    ServiceBusException serviceBusException = (ServiceBusException) throwable;
    return serviceBusException.getIsTransient();
  }
  return false;
}
 

import com.microsoft.azure.servicebus.ServiceBusException; //导入依赖的package包/类
@Override
public void init() {
  String remoteHost = String.format(EVENTHUB_REMOTE_HOST_FORMAT, eventHubNamespace);
  try {
    ConnectionStringBuilder connectionStringBuilder =
            new ConnectionStringBuilder(eventHubNamespace, entityPath, sasKeyName, sasKey);

    eventHubClient = EventHubClient.createFromConnectionStringSync(connectionStringBuilder.toString(), retryPolicy);
  } catch (IOException | ServiceBusException e) {
    String msg = String.format("Creation of EventHub client failed for eventHub EntityPath: %s on remote host %s:%d",
            entityPath, remoteHost, ClientConstants.AMQPS_PORT);
    LOG.error(msg, e);
    throw new SamzaException(msg, e);
  }
}
 

import com.microsoft.azure.servicebus.ServiceBusException; //导入依赖的package包/类
@Test
public void testReceive() throws ServiceBusException {
  EventHubClientManagerFactory clientFactory = new EventHubClientManagerFactory();
  EventHubClientManager wrapper = clientFactory
          .getEventHubClientManager(SYSTEM_NAME, STREAM_NAME1, new EventHubConfig(createEventHubConfig()));
  wrapper.init();
  EventHubClient client = wrapper.getEventHubClient();
  PartitionReceiver receiver =
          client.createReceiverSync(EventHubClient.DEFAULT_CONSUMER_GROUP_NAME, "0",
                  EventHubSystemConsumer.START_OF_STREAM, true);
  receiveMessages(receiver, 300);
}
 

import com.microsoft.azure.servicebus.ServiceBusException; //导入依赖的package包/类
private void receiveMessages(PartitionReceiver receiver, int numMessages) throws ServiceBusException {
  int count = 0;
  while (count < numMessages) {

    Iterable<EventData> messages = receiver.receiveSync(100);
    if (messages == null) {
      break;
    }
    for (EventData data : messages) {
      count++;
      LOG.info("Data" + new String(data.getBytes()));
    }
  }
}
 

import com.microsoft.azure.servicebus.ServiceBusException; //导入依赖的package包/类
public EventHubClient createEventHubClient() throws IOException, ServiceBusException {
  ConnectionStringBuilder connStr = new ConnectionStringBuilder(
      commonConf.namespaceName,
      commonConf.eventHubName,
      commonConf.sasKeyName,
      commonConf.sasKey
  );
  return EventHubClient.createFromConnectionStringSync(connStr.toString());
}