我有以下问题。
我们使用事件中心。在下面的类中,我们从 IEventProcessor 继承,如您所见,我们使用 Service Locator 。我们不能让它与构造函数/属性注入(inject)一起工作。当旨在从 IEventProcessor 继承的类时,CaSTLe Windsor 似乎无法解决依赖关系。这是一个已知问题还是我需要做些什么才能让它工作?
下面是代码:
public class EventProcessor : IEventProcessor
{
private readonly IEventService _eventService;
private readonly ILogger _logger;
private readonly Lazy<RetryPolicy> _retryPolicy;
private readonly IConfigurationProvider _configurationProvider;
public EventProcessor()
{
try
{
_eventService = ContainerProvider.Current.Container.Resolve<IEventService>();
_logger = ContainerProvider.Current.Container.Resolve<ILogger>();
_configurationProvider = ContainerProvider.Current.Container.Resolve<IConfigurationProvider>();
}
catch (Exception exception)
{
_logger.WriteError(string.Format("Error occured when intializing EventProcessor: '{0}'", exception));
}
}
public Task OpenAsync(PartitionContext context)
{
return Task.FromResult(0);
}
public async Task ProcessEventsAsync(PartitionContext context, IEnumerable<EventData> events)
{
var eventsList = events.ToList();
EventData lastEvent = null;
foreach (var eventData in eventsList)
{
_logger.WriteVerbose(string.Format("Consumming {0} events...", eventsList.Count()));
_eventService.ProcessEvent(eventData);
lastEvent = eventData;
}
if (lastEvent != null)
{
await AzureServiceBusRetryPolicy.ExecuteAsync(async () => await context.CheckpointAsync(lastEvent));
}
}
public async Task CloseAsync(PartitionContext context, CloseReason reason)
{
_logger.WriteInfo("EventHub processor was closed for this reason: " + reason);
if (reason == CloseReason.Shutdown)
{
await AzureServiceBusRetryPolicy.ExecuteAsync(async () => await context.CheckpointAsync());
}
}
}
谢谢
最佳答案
我正在使用 Autofac,但遇到了同样的问题。
我通过实现 IEventProcessorFactory 并在 EventProcessorHost 中注册处理器时使用它来解决。
举个例子,我的 EventProcessorHost
看起来像这样:
public class EventHubProcessorHost
{
private readonly IEventProcessorFactory _eventProcessorFactory;
private readonly string _serviceBusConnectionString;
private readonly string _storageConnectionString;
private readonly string _eventHubName;
public EventHubProcessorHost(IEventProcessorFactory eventProcessorFactory, string serviceBusConnectionString, string storageConnectionString, string eventHubName)
{
_eventProcessorFactory = eventProcessorFactory;
_serviceBusConnectionString = serviceBusConnectionString;
_storageConnectionString = storageConnectionString;
_eventHubName = eventHubName;
}
public void Start()
{
var builder = new ServiceBusConnectionStringBuilder(_serviceBusConnectionString)
{
TransportType = TransportType.Amqp
};
var client = EventHubClient.CreateFromConnectionString(builder.ToString(), _eventHubName);
try
{
var eventProcessorHost = new EventProcessorHost("singleworker",
client.Path, client.GetDefaultConsumerGroup().GroupName, builder.ToString(), _storageConnectionString);
eventProcessorHost.RegisterEventProcessorFactoryAsync(_eventProcessorFactory);
}
catch (Exception exp)
{
Console.WriteLine("Error on send: " + exp.Message);
}
}
}
我传入的工厂引用了我的 IoC 容器:
public class MyEventProcessorFactory : IEventProcessorFactory
{
private readonly IComponentContext _componentContext;
public MyEventProcessorFactory(IComponentContext componentContext)
{
_componentContext = componentContext;
}
public IEventProcessor CreateEventProcessor(PartitionContext context)
{
return _componentContext.Resolve<IEventProcessor>();
}
}
这允许我在
EventProcessor
中正常使用构造函数注入(inject):public class MyEventProcessor : IEventProcessor
{
private IFoo _foo;
public MyEventProcessor(IFoo foo)
{
_foo = foo;
}
public Task OpenAsync(PartitionContext context)
{
return Task.FromResult<object>(null);
}
public async Task ProcessEventsAsync(PartitionContext context, IEnumerable<EventData> events)
{
foreach (var eventData in events)
{
// Processing code
}
await context.CheckpointAsync();
}
public async Task CloseAsync(PartitionContext context, CloseReason reason)
{
if (reason == CloseReason.Shutdown)
{
await context.CheckpointAsync();
}
}
}
然后我只是在启动时在 Autofac 容器中正常连接所有东西:
builder.RegisterType<Foo>().As<IFoo>()
builder.RegisterType<MyEventProcessor>().As<IEventProcessor>()
builder.Register(c => new MyEventProcessorFactory(c.Resolve<IComponentContext>())).As<IEventProcessorFactory>();
希望这可以帮助。
关于caSTLe-windsor - 如何将依赖项传递给实现 IEventProcessor(Event Hub) 的类,我们在Stack Overflow上找到一个类似的问题:https://stackoverflow.com/questions/27849241/