Azure Event Hub 技术研究系列3-Event Hub接收事件
上篇博文中，我们通过编程的方式介绍了如何将事件消息发送到Azure Event Hub：Azure Event Hub 技术研究系列2-发送事件到Event Hub
本篇文章中，我们继续：从Event Hub中接收事件。
1. 新建控制台工程 EventHubReceiver
2. 添加Nuget引用
Microsoft.Azure.EventHubs
Microsoft.Azure.EventHubs.Processor
3. 实现IEventProcessor接口
MyEventProcessor
1 using Microsoft.Azure.EventHubs;
2 using Microsoft.Azure.EventHubs.Processor;
3 using System.Threading.Tasks;
4
5 public class MyEventProcessor : IEventProcessor
6 {
7 public Task CloseAsync(PartitionContext context, CloseReason reason)
8 {
9 Console.WriteLine($"Processor Shutting Down. Partition '{context.PartitionId}', Reason: '{reason}'.");
10 return Task.CompletedTask;
11 }
12
13 public Task OpenAsync(PartitionContext context)
14 {
15 Console.WriteLine($"MyEventProcessor initialized. Partition: '{context.PartitionId}'");
16 return Task.CompletedTask;
17 }
18
19 public Task ProcessErrorAsync(PartitionContext context, Exception error)
20 {
21 Console.WriteLine($"Error on Partition: {context.PartitionId}, Error: {error.Message}");
22 return Task.CompletedTask;
23 }
24
25 public Task ProcessEventsAsync(PartitionContext context, IEnumerable<EventData> messages)
26 {
27 foreach (var eventData in messages)
28 {
29 var data = Encoding.UTF8.GetString(eventData.Body.Array, eventData.Body.Offset, eventData.Body.Count);
30 Console.WriteLine($"Event message received. Partition: '{context.PartitionId}', Data: '{data}'");
31 }
32
33 return context.CheckpointAsync();
34 }
35 }
4. Program程序
添加常量作为事件中心连接字符串、事件中心名称、存储帐户容器名称、存储帐户名称和存储帐户密钥。 添加以下代码，并将占位符替换为其对应的值。
private const string EhConnectionString = "{Event Hubs connection string}";
private const string EhEntityPath = "{Event Hub path/name}"; //MyEventHub
private const string StorageContainerName = "{Storage account container name}"; //eventhubcontainer
private const string StorageAccountName = "{Storage account name}"; //linux1
private const string StorageAccountKey = "{Storage account key}";
private static readonly string StorageConnectionString = string.Format("DefaultEndpointsProtocol=https;AccountName={0};AccountKey={1}", StorageAccountName, StorageAccountKey);
这里涉及到Azure Storage Account，必须为上篇博文中创建的事件中心MyEventHub指定一个存储账户和存储容器
增加MainAysnc方法：注册事件处理器，处理事件消息
1 /// <summary>
2 /// 注册事件处理器
3 /// </summary>
4 /// <param name="args"></param>
5 /// <returns></returns>
6 private static async Task MainAsync(string[] args)
7 {
8 Console.WriteLine("Registering EventProcessor...");
9
10 var eventProcessorHost = new EventProcessorHost(
11 EhEntityPath,
12 PartitionReceiver.DefaultConsumerGroupName,
13 EhConnectionString,
14 StorageConnectionString,
15 StorageContainerName);
16
17 // Registers the Event Processor Host and starts receiving messages
18 await eventProcessorHost.RegisterEventProcessorAsync<MyEventProcessor>();
19
20 Console.WriteLine("Receiving. Press ENTER to stop worker.");
21 Console.ReadLine();
22
23 // Disposes of the Event Processor Host
24 await eventProcessorHost.UnregisterEventProcessorAsync();
25 }
Main函数
1 static void Main(string[] args)
2 {
3 MainAsync(args).GetAwaiter().GetResult();
4 }
Run
至此，我们实现了事件消息发送到Event Hub，同时从Event Hub接收处理事件消息。
周国庆
2017/5/18
页:
[1]