Azure Event hub usage
2024-08-30 06:35:13
1. create event hub on azure
watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQv/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/Center" alt="" />
2. create a prj , event hub sender, install nuget pkg - azure service bus
3. check connection string
4. sender sample code
static void Main(string[] args)
{
Console.WriteLine("Press Ctrl-C to stop the sender process");
Console.WriteLine("Press Enter to start now");
Console.ReadLine();
SendingRandomMessages();
} static string eventHubName = "get from event hub connection information";
static string connectionString = "get from event hub connection information"; static void SendingRandomMessages()
{
var eventHubClient = EventHubClient.CreateFromConnectionString(connectionString, eventHubName);
while (true)
{
try
{
var message = Guid.NewGuid().ToString();
Console.WriteLine("{0} > Sending message: {1}", DateTime.Now, message);
eventHubClient.Send(new EventData(Encoding.UTF8.GetBytes(message)));
}
catch (Exception exception)
{
Console.ForegroundColor = ConsoleColor.Red;
Console.WriteLine("{0} > Exception: {1}", DateTime.Now, exception.Message);
Console.ResetColor();
} Thread.Sleep(200);
}
}
5. create a storage account
watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQv/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/Center" alt="" />
6. install nuget for receiver prj
7. check keys
8. sample code for receiver
class SimpleEventProcessor : IEventProcessor
{
Stopwatch checkpointStopWatch; async Task IEventProcessor.CloseAsync(PartitionContext context, CloseReason reason)
{
Console.WriteLine("Processor Shutting Down. Partition '{0}', Reason: '{1}'.", context.Lease.PartitionId, reason);
if (reason == CloseReason.Shutdown)
{
await context.CheckpointAsync();
}
} Task IEventProcessor.OpenAsync(PartitionContext context)
{
Console.WriteLine("SimpleEventProcessor initialized. Partition: '{0}', Offset: '{1}'", context.Lease.PartitionId, context.Lease.Offset);
this.checkpointStopWatch = new Stopwatch();
this.checkpointStopWatch.Start();
return Task.FromResult<object>(null);
} async Task IEventProcessor.ProcessEventsAsync(PartitionContext context, IEnumerable<EventData> messages)
{
foreach (EventData eventData in messages)
{
string data = Encoding.UTF8.GetString(eventData.GetBytes()); Console.WriteLine(string.Format("Message received. Partition: '{0}', Data: '{1}'",
context.Lease.PartitionId, data));
} //Call checkpoint every 5 minutes, so that worker can resume processing from 5 minutes back if it restarts.
if (this.checkpointStopWatch.Elapsed > TimeSpan.FromMinutes(5))
{
await context.CheckpointAsync();
this.checkpointStopWatch.Restart();
}
}
} static void Main(string[] args)
{
string eventHubConnectionString = "get from azure event hub connection information";
string eventHubName = "get from azure event hub connection information";
string storageAccountName = "get from azure storage keys";
string storageAccountKey = "get from azure storage keys";
string storageConnectionString = string.Format("DefaultEndpointsProtocol=https;AccountName={0};AccountKey={1}", storageAccountName, storageAccountKey); string eventProcessorHostName = Guid.NewGuid().ToString();
EventProcessorHost eventProcessorHost = new EventProcessorHost(eventProcessorHostName, eventHubName, EventHubConsumerGroup.DefaultGroupName, eventHubConnectionString, storageConnectionString);
Console.WriteLine("Registering EventProcessor...");
var options = new EventProcessorOptions();
options.ExceptionReceived += (sender, e) => { Console.WriteLine(e.Exception); };
eventProcessorHost.RegisterEventProcessorAsync<SimpleEventProcessor>(options).Wait(); Console.WriteLine("Receiving. Press enter key to stop worker.");
Console.ReadLine();
eventProcessorHost.UnregisterEventProcessorAsync().Wait();
}
watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQv/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/Center" alt="" />
最新文章
- 【原】移动web资源整理
- LR常用函数以及调用自定义函数
- 3d加速的一些问题
- 002 C#学前入门
- poj 2763: [JLOI2011]飞行路线(spfa分层图最短路)
- bzoj 2243 [SDOI2011]染色(树链剖分,线段树)
- 关于USACO Training
- Swift的基础,操作符,字符串和集合类型
- BZOJ 3237([Ahoi2013]连通图-cdq图重构-连通性缩点)
- SQL入门学习0-数据库与SQL
- 使用Windows2003创建DHCP服务器 - 进阶者系列 - 学习者系列文章
- USACO 3.3 Riding the Fences
- Java高并发如何解决
- Linux命令行总结
- 17-Flink消费Kafka写入Mysql
- SpringSecurityOAuth认证配置及Token的存储
- Spark2 Dataset去重、差集、交集
- 谈一谈chrome浏览器使用
- JAVA学习6:用Maven创建Spring3 MVC项目
- Thread-Specific-Storage for C/C++