2

我希望获得有关如何将 EventProcessorHost 与辅助角色一起使用的指导。基本上我希望让 EventProcessorHost 并行处理分区，我想知道我应该在哪里将这种类型的代码放置在工作角色中，以及我是否缺少任何关键。

    var manager = NamespaceManager.CreateFromConnectionString(connectionString);
    var desc = manager.CreateEventHubIfNotExistsAsync(path).Result;
    var client = Microsoft.ServiceBus.Messaging.EventHubClient.CreateFromConnectionString(connectionString, path);
    var host = new EventProcessorHost(hostname, path, consumerGroup, connectionString, blobStorageConnectionString);
    EventHubProcessorFactory<EventData> factory = new EventHubProcessorFactory<EventData>();
    host.RegisterEventProcessorFactoryAsync(factory);

我读过的所有内容都说 EventProcessorHost 会自行划分分区，但是上面的代码足以异步处理所有分区吗？

4

1 回答 1

3

这是我们如何从工作角色处理事件中心的简化版本。我们将实例保留在 mainWorker 角色中并调用 IEventProcessor 开始处理它。

这样我们就可以在 Worker 响应关闭事件等时调用它并关闭它。

编辑：

至于并行处理，当 IEventProcessor 类完成当前事件的处理后，它只会从事件中心获取 10 个以上的事件。为您处理所有精美的隔断租赁。

这是一个同步工作流，当我扩展到多个工作角色时，我开始看到分区在实例之间拆分并且变得更快等。如果您希望它以不同的方式处理事件中心，则必须推出自己的解决方案.

public class WorkerRole : RoleEntryPoint
{
    private readonly CancellationTokenSource _cancellationTokenSource = new CancellationTokenSource();
    private readonly ManualResetEvent _runCompleteEvent = new ManualResetEvent(false);
    private EventProcessorHost _eventProcessorHost;

     public override bool OnStart()
    {
        ThreadPool.SetMaxThreads(4096, 2048);
        ServicePointManager.DefaultConnectionLimit = 500;
        ServicePointManager.UseNagleAlgorithm = false;
        ServicePointManager.Expect100Continue = false;

        var eventClient = EventHubClient.CreateFromConnectionString("consumersConnectionString",
                            "eventHubName");
        _eventProcessorHost = new EventProcessorHost(Dns.GetHostName(), eventClient.Path, 
               eventClient.GetDefaultConsumerGroup().GroupName,
               "consumersConnectionString", "blobLeaseConnectionString");
        return base.OnStart();
    }

    public override void Run()
    {
        try
        {
            RunAsync(this._cancellationTokenSource.Token).Wait();
        }
        finally
        {
            _runCompleteEvent.Set();
        }
    }

    private async Task RunAsync(CancellationToken cancellationToken)
    {
        // starts processing here
        await _eventProcessorHost.RegisterEventProcessorAsync<EventProcessor>();
        while (!cancellationToken.IsCancellationRequested)
        {
            await Task.Delay(TimeSpan.FromMinutes(1));
        }
    }

    public override void OnStop()
    {
        _eventProcessorHost.UnregisterEventProcessorAsync().Wait();
        _cancellationTokenSource.Cancel();
        _runCompleteEvent.WaitOne();
        base.OnStop();
    }
}

我有用于特定分区的多个处理器（您可以通过这种方式保证 FIFO），但您可以轻松实现自己的逻辑，即在我的示例中跳过使用 EventDataProcessor 类和 Dictionary 查找，只需在 ProcessEventsAsync 方法中实现一些逻辑.

public class EventProcessor : IEventProcessor
{
    private readonly Dictionary<string, IEventDataProcessor> _eventDataProcessors; 

    public EventProcessor()
    {
        _eventDataProcessors = new Dictionary<string, IEventDataProcessor>
        {
            {"A", new EventDataProcessorA()},
            {"B", new EventDataProcessorB()},
            {"C", new EventDataProcessorC()}
        }
    }

    public Task OpenAsync(PartitionContext context)
    {
        return Task.FromResult<object>(null);
    }

    public async Task ProcessEventsAsync(PartitionContext context, IEnumerable<EventData> messages)
    {
        foreach(EventData eventData in messages)
        {
            // implement your own logic here, you could just process the data here, just remember that they will all be from the same partition in this block
            try
            {
                IEventDataProcessor eventDataProcessor;
                if(_eventDataProcessors.TryGetValue(eventData.PartitionKey, out eventDataProcessor))
                {
                    await eventDataProcessor.ProcessMessage(eventData);
                }
            }
            catch (Exception ex)
            {
                _//log exception
            }
        }
        await context.CheckpointAsync();
    }

    public async Task CloseAsync(PartitionContext context, CloseReason reason)
    {
        if (reason == CloseReason.Shutdown)
            await context.CheckpointAsync();
    }
}

我们的 EventDataProcessor 之一的示例

public interface IEventDataProcessor
{
    Task ProcessMessage(EventData eventData);
}

public class EventDataProcessorA : IEventDataProcessor
{
    public async Task ProcessMessage(EventData eventData)
    {
        // Do Something specific with data from Partition "A"
    }
}

public class EventDataProcessorB : IEventDataProcessor
{
    public async Task ProcessMessage(EventData eventData)
    {
        // Do Something specific with data from Partition "B"
    }
}

希望这会有所帮助，到目前为止，它对我们来说一直坚如磐石，并且可以轻松扩展到多个实例

于 2015-06-01T19:05:50.160 回答