Azure Event Hub 技术研究系列3-Event Hub接收事件

sszxf

　　上篇博文中，我们通过编程的方式介绍了如何将事件消息发送到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