import com.microsoft.azure.eventhubs.EventHubClient; //导入依赖的package包/类
private CompletableFuture<Void> sendToEventHub(String streamId, EventData eventData, Object partitionKey,
                                               EventHubClient eventHubClient) {
  if (PartitioningMethod.ROUND_ROBIN.equals(partitioningMethod)) {
    return eventHubClient.send(eventData);
  } else if (PartitioningMethod.EVENT_HUB_HASHING.equals(partitioningMethod)) {
    if (partitionKey == null) {
      throw new SamzaException("Partition key cannot be null for EventHub hashing");
    }
    return eventHubClient.send(eventData, convertPartitionKeyToString(partitionKey));
  } else if (PartitioningMethod.PARTITION_KEY_AS_PARTITION.equals(partitioningMethod)) {
    if (!(partitionKey instanceof Integer)) {
      String msg = "Partition key should be of type Integer";
      throw new SamzaException(msg);
    }

    Integer numPartition = streamPartitionSenders.get(streamId).size();
    Integer destinationPartition = (Integer) partitionKey % numPartition;

    PartitionSender sender = streamPartitionSenders.get(streamId).get(destinationPartition);
    return sender.send(eventData);
  } else {
    throw new SamzaException("Unknown partitioning method " + partitioningMethod);
  }
}
 

import com.microsoft.azure.eventhubs.EventHubClient; //导入依赖的package包/类
/**
 * No-args constructor.
 */
public AzureIoTService() {
	if (!StringUtils.isBlank(SERVICE_ACCESS_KEY)) {
		if (!StringUtils.isBlank(EVENTHUB_ENDPOINT)) {
			final ArrayList<EventHubClient> recievers = new ArrayList<>();
			for (int i = 0; i < PARTITIONS_COUNT; i++) {
				recievers.add(receiveEventsAsync(Integer.toString(i)));
			}
			Para.addDestroyListener(new DestroyListener() {
				public void onDestroy() {
					for (EventHubClient recvr : recievers) {
						recvr.close();
					}
				}
			});
		}
		try {
			registryManager = RegistryManager.createFromConnectionString(SERVICE_CONN_STR);
		} catch (Exception ex) {
			logger.warn("Couldn't initialize Azure registry manager: {}", ex.getMessage());
		}
	}
}
 

import com.microsoft.azure.eventhubs.EventHubClient; //导入依赖的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.eventhubs.EventHubClient; //导入依赖的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.eventhubs.EventHubClient; //导入依赖的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.eventhubs.EventHubClient; //导入依赖的package包/类
public EventHubClient createEventHubClient() throws IOException, ServiceBusException {
  ConnectionStringBuilder connStr = new ConnectionStringBuilder(
      commonConf.namespaceName,
      commonConf.eventHubName,
      commonConf.sasKeyName,
      commonConf.sasKey
  );
  return EventHubClient.createFromConnectionStringSync(connStr.toString());
}
 

import com.microsoft.azure.eventhubs.EventHubClient; //导入依赖的package包/类
@Override
public List<ConfigIssue> init(Info info, Context context) {
  List<ConfigIssue> issues = new ArrayList<>();
  this.context = context;

  consumerConfigBean.dataFormatConfig.stringBuilderPoolSize = getNumberOfThreads();
  consumerConfigBean.dataFormatConfig.init(
      context,
      consumerConfigBean.dataFormat,
      Groups.DATA_FORMAT.name(),
      "dataFormatConfig",
      DataFormatConstants.MAX_OVERRUN_LIMIT,
      issues
  );
  parserFactory = consumerConfigBean.dataFormatConfig.getParserFactory();
  errorQueue = new ArrayBlockingQueue<>(100);
  errorList = new ArrayList<>(100);

  // validate the connection info
  if(issues.size() == 0) {
    try {
      EventHubClient ehClient = eventHubCommon.createEventHubClient();
      EventHubRuntimeInformation ehInfo = ehClient.getRuntimeInformation().get();
      ehClient.close().get();
    } catch (Exception e) {
      issues.add(context.createConfigIssue(
          Groups.EVENT_HUB.toString(),
          EventHubCommon.CONF_NAME_SPACE,
          Errors.EVENT_HUB_02,
          e.getMessage()
      ));
    }
  }

  return issues;
}
 

import com.microsoft.azure.eventhubs.EventHubClient; //导入依赖的package包/类
private static EventHubClient receiveEventsAsync(final String partitionId) {
	EventHubClient client = null;
	try {
		client = EventHubClient.createFromConnectionStringSync(EVENTHUB_CONN_STR);
		client.createReceiver(EventHubClient.DEFAULT_CONSUMER_GROUP_NAME, partitionId, Instant.now()).
				thenAccept(new Receiver(partitionId));
	} catch (Exception e) {
		logger.warn("Couldn't start receiving messages from Azure cloud: {}", e.getMessage());
	}
	return client;
}
 

import com.microsoft.azure.eventhubs.EventHubClient; //导入依赖的package包/类
@Override
public EventHubClient getEventHubClient() {
  return eventHubClient;
}
 

import com.microsoft.azure.eventhubs.EventHubClient; //导入依赖的package包/类
/**
 * Returns the underlying {@link EventHubClient} instance. Multiple invocations
 * of this method should return the same instance instead of
 * creating new ones.
 *
 * @return EventHub client instance of the wrapper
 */
EventHubClient getEventHubClient();