mirror of https://github.com/abpframework/abp.git
Halil İbrahim Kalkan
6 years ago
committed by
GitHub
GitHub
24 changed files with 937 additions and 0 deletions
@ -0,0 +1,3 @@ |
|||
<Weavers xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="FodyWeavers.xsd"> |
|||
<ConfigureAwait ContinueOnCapturedContext="false" /> |
|||
</Weavers> |
|||
@ -0,0 +1,30 @@ |
|||
<?xml version="1.0" encoding="utf-8"?> |
|||
<xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema"> |
|||
<!-- This file was generated by Fody. Manual changes to this file will be lost when your project is rebuilt. --> |
|||
<xs:element name="Weavers"> |
|||
<xs:complexType> |
|||
<xs:all> |
|||
<xs:element name="ConfigureAwait" minOccurs="0" maxOccurs="1"> |
|||
<xs:complexType> |
|||
<xs:attribute name="ContinueOnCapturedContext" type="xs:boolean" /> |
|||
</xs:complexType> |
|||
</xs:element> |
|||
</xs:all> |
|||
<xs:attribute name="VerifyAssembly" type="xs:boolean"> |
|||
<xs:annotation> |
|||
<xs:documentation>'true' to run assembly verification (PEVerify) on the target assembly after all weavers have been executed.</xs:documentation> |
|||
</xs:annotation> |
|||
</xs:attribute> |
|||
<xs:attribute name="VerifyIgnoreCodes" type="xs:string"> |
|||
<xs:annotation> |
|||
<xs:documentation>A comma-separated list of error codes that can be safely ignored in assembly verification.</xs:documentation> |
|||
</xs:annotation> |
|||
</xs:attribute> |
|||
<xs:attribute name="GenerateXsd" type="xs:boolean"> |
|||
<xs:annotation> |
|||
<xs:documentation>'false' to turn off automatic generation of the XML Schema file.</xs:documentation> |
|||
</xs:annotation> |
|||
</xs:attribute> |
|||
</xs:complexType> |
|||
</xs:element> |
|||
</xs:schema> |
|||
@ -0,0 +1,16 @@ |
|||
<Project Sdk="Microsoft.NET.Sdk"> |
|||
|
|||
<Import Project="..\..\..\configureawait.props" /> |
|||
<Import Project="..\..\..\common.props" /> |
|||
|
|||
<PropertyGroup> |
|||
<TargetFramework>netstandard2.0</TargetFramework> |
|||
<RootNamespace /> |
|||
</PropertyGroup> |
|||
|
|||
<ItemGroup> |
|||
<ProjectReference Include="..\Volo.Abp.EventBus\Volo.Abp.EventBus.csproj" /> |
|||
<ProjectReference Include="..\Volo.Abp.Kafka\Volo.Abp.Kafka.csproj" /> |
|||
</ItemGroup> |
|||
|
|||
</Project> |
|||
@ -0,0 +1,27 @@ |
|||
using Microsoft.Extensions.DependencyInjection; |
|||
using Volo.Abp.Kafka; |
|||
using Volo.Abp.Modularity; |
|||
|
|||
namespace Volo.Abp.EventBus.Kafka |
|||
{ |
|||
[DependsOn( |
|||
typeof(AbpEventBusModule), |
|||
typeof(AbpKafkaModule))] |
|||
public class AbpEventBusKafkaModule : AbpModule |
|||
{ |
|||
public override void ConfigureServices(ServiceConfigurationContext context) |
|||
{ |
|||
var configuration = context.Services.GetConfiguration(); |
|||
|
|||
Configure<AbpKafkaEventBusOptions>(configuration.GetSection("Kafka:EventBus")); |
|||
} |
|||
|
|||
public override void OnApplicationInitialization(ApplicationInitializationContext context) |
|||
{ |
|||
context |
|||
.ServiceProvider |
|||
.GetRequiredService<KafkaDistributedEventBus>() |
|||
.Initialize(); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,12 @@ |
|||
namespace Volo.Abp.EventBus.Kafka |
|||
{ |
|||
public class AbpKafkaEventBusOptions |
|||
{ |
|||
|
|||
public string ConnectionName { get; set; } |
|||
|
|||
public string TopicName { get; set; } |
|||
|
|||
public string GroupId { get; set; } |
|||
} |
|||
} |
|||
@ -0,0 +1,208 @@ |
|||
using System; |
|||
using System.Collections.Concurrent; |
|||
using System.Collections.Generic; |
|||
using System.Linq; |
|||
using System.Threading.Tasks; |
|||
using Confluent.Kafka; |
|||
using Microsoft.Extensions.DependencyInjection; |
|||
using Microsoft.Extensions.Options; |
|||
using Volo.Abp.DependencyInjection; |
|||
using Volo.Abp.EventBus.Distributed; |
|||
using Volo.Abp.Kafka; |
|||
using Volo.Abp.MultiTenancy; |
|||
using Volo.Abp.Threading; |
|||
|
|||
namespace Volo.Abp.EventBus.Kafka |
|||
{ |
|||
[Dependency(ReplaceServices = true)] |
|||
[ExposeServices(typeof(IDistributedEventBus), typeof(KafkaDistributedEventBus))] |
|||
public class KafkaDistributedEventBus : EventBusBase, IDistributedEventBus, ISingletonDependency |
|||
{ |
|||
protected AbpKafkaEventBusOptions AbpKafkaEventBusOptions { get; } |
|||
protected AbpDistributedEventBusOptions AbpDistributedEventBusOptions { get; } |
|||
protected IKafkaMessageConsumerFactory MessageConsumerFactory { get; } |
|||
protected IKafkaSerializer Serializer { get; } |
|||
protected IProducerPool ProducerPool { get; } |
|||
protected ConcurrentDictionary<Type, List<IEventHandlerFactory>> HandlerFactories { get; } |
|||
protected ConcurrentDictionary<string, Type> EventTypes { get; } |
|||
protected IKafkaMessageConsumer Consumer { get; private set; } |
|||
|
|||
public KafkaDistributedEventBus( |
|||
IServiceScopeFactory serviceScopeFactory, |
|||
ICurrentTenant currentTenant, |
|||
IOptions<AbpKafkaEventBusOptions> abpKafkaEventBusOptions, |
|||
IKafkaMessageConsumerFactory messageConsumerFactory, |
|||
IOptions<AbpDistributedEventBusOptions> abpDistributedEventBusOptions, |
|||
IKafkaSerializer serializer, |
|||
IProducerPool producerPool) |
|||
: base(serviceScopeFactory, currentTenant) |
|||
{ |
|||
AbpKafkaEventBusOptions = abpKafkaEventBusOptions.Value; |
|||
AbpDistributedEventBusOptions = abpDistributedEventBusOptions.Value; |
|||
MessageConsumerFactory = messageConsumerFactory; |
|||
Serializer = serializer; |
|||
ProducerPool = producerPool; |
|||
|
|||
HandlerFactories = new ConcurrentDictionary<Type, List<IEventHandlerFactory>>(); |
|||
EventTypes = new ConcurrentDictionary<string, Type>(); |
|||
} |
|||
|
|||
public void Initialize() |
|||
{ |
|||
Consumer = MessageConsumerFactory.Create( |
|||
AbpKafkaEventBusOptions.TopicName, |
|||
AbpKafkaEventBusOptions.GroupId, |
|||
AbpKafkaEventBusOptions.ConnectionName); |
|||
|
|||
Consumer.OnMessageReceived(ProcessEventAsync); |
|||
|
|||
SubscribeHandlers(AbpDistributedEventBusOptions.Handlers); |
|||
} |
|||
|
|||
private async Task ProcessEventAsync(Message<string, byte[]> message) |
|||
{ |
|||
var eventName = message.Key; |
|||
var eventType = EventTypes.GetOrDefault(eventName); |
|||
if (eventType == null) |
|||
{ |
|||
return; |
|||
} |
|||
|
|||
var eventData = Serializer.Deserialize(message.Value, eventType); |
|||
|
|||
await TriggerHandlersAsync(eventType, eventData); |
|||
} |
|||
|
|||
public IDisposable Subscribe<TEvent>(IDistributedEventHandler<TEvent> handler) where TEvent : class |
|||
{ |
|||
return Subscribe(typeof(TEvent), handler); |
|||
} |
|||
|
|||
public override IDisposable Subscribe(Type eventType, IEventHandlerFactory factory) |
|||
{ |
|||
var handlerFactories = GetOrCreateHandlerFactories(eventType); |
|||
|
|||
if (factory.IsInFactories(handlerFactories)) |
|||
{ |
|||
return NullDisposable.Instance; |
|||
} |
|||
|
|||
handlerFactories.Add(factory); |
|||
|
|||
return new EventHandlerFactoryUnregistrar(this, eventType, factory); |
|||
} |
|||
|
|||
/// <inheritdoc/>
|
|||
public override void Unsubscribe<TEvent>(Func<TEvent, Task> action) |
|||
{ |
|||
Check.NotNull(action, nameof(action)); |
|||
|
|||
GetOrCreateHandlerFactories(typeof(TEvent)) |
|||
.Locking(factories => |
|||
{ |
|||
factories.RemoveAll( |
|||
factory => |
|||
{ |
|||
var singleInstanceFactory = factory as SingleInstanceHandlerFactory; |
|||
if (singleInstanceFactory == null) |
|||
{ |
|||
return false; |
|||
} |
|||
|
|||
var actionHandler = singleInstanceFactory.HandlerInstance as ActionEventHandler<TEvent>; |
|||
if (actionHandler == null) |
|||
{ |
|||
return false; |
|||
} |
|||
|
|||
return actionHandler.Action == action; |
|||
}); |
|||
}); |
|||
} |
|||
|
|||
/// <inheritdoc/>
|
|||
public override void Unsubscribe(Type eventType, IEventHandler handler) |
|||
{ |
|||
GetOrCreateHandlerFactories(eventType) |
|||
.Locking(factories => |
|||
{ |
|||
factories.RemoveAll( |
|||
factory => |
|||
factory is SingleInstanceHandlerFactory handlerFactory && |
|||
handlerFactory.HandlerInstance == handler |
|||
); |
|||
}); |
|||
} |
|||
|
|||
/// <inheritdoc/>
|
|||
public override void Unsubscribe(Type eventType, IEventHandlerFactory factory) |
|||
{ |
|||
GetOrCreateHandlerFactories(eventType).Locking(factories => factories.Remove(factory)); |
|||
} |
|||
|
|||
/// <inheritdoc/>
|
|||
public override void UnsubscribeAll(Type eventType) |
|||
{ |
|||
GetOrCreateHandlerFactories(eventType).Locking(factories => factories.Clear()); |
|||
} |
|||
|
|||
public override async Task PublishAsync(Type eventType, object eventData) |
|||
{ |
|||
var eventName = EventNameAttribute.GetNameOrDefault(eventType); |
|||
var body = Serializer.Serialize(eventData); |
|||
|
|||
var producer = ProducerPool.Get(AbpKafkaEventBusOptions.ConnectionName); |
|||
|
|||
await producer.ProduceAsync( |
|||
AbpKafkaEventBusOptions.TopicName, |
|||
new Message<string, byte[]> |
|||
{ |
|||
Key = eventName, Value = body |
|||
}); |
|||
} |
|||
|
|||
private List<IEventHandlerFactory> GetOrCreateHandlerFactories(Type eventType) |
|||
{ |
|||
return HandlerFactories.GetOrAdd( |
|||
eventType, |
|||
type => |
|||
{ |
|||
var eventName = EventNameAttribute.GetNameOrDefault(type); |
|||
EventTypes[eventName] = type; |
|||
return new List<IEventHandlerFactory>(); |
|||
} |
|||
); |
|||
} |
|||
|
|||
protected override IEnumerable<EventTypeWithEventHandlerFactories> GetHandlerFactories(Type eventType) |
|||
{ |
|||
var handlerFactoryList = new List<EventTypeWithEventHandlerFactories>(); |
|||
|
|||
foreach (var handlerFactory in HandlerFactories.Where(hf => ShouldTriggerEventForHandler(eventType, hf.Key)) |
|||
) |
|||
{ |
|||
handlerFactoryList.Add( |
|||
new EventTypeWithEventHandlerFactories(handlerFactory.Key, handlerFactory.Value)); |
|||
} |
|||
|
|||
return handlerFactoryList.ToArray(); |
|||
} |
|||
|
|||
private static bool ShouldTriggerEventForHandler(Type targetEventType, Type handlerEventType) |
|||
{ |
|||
//Should trigger same type
|
|||
if (handlerEventType == targetEventType) |
|||
{ |
|||
return true; |
|||
} |
|||
|
|||
//Should trigger for inherited types
|
|||
if (handlerEventType.IsAssignableFrom(targetEventType)) |
|||
{ |
|||
return true; |
|||
} |
|||
|
|||
return false; |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,3 @@ |
|||
<Weavers xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="FodyWeavers.xsd"> |
|||
<ConfigureAwait ContinueOnCapturedContext="false" /> |
|||
</Weavers> |
|||
@ -0,0 +1,30 @@ |
|||
<?xml version="1.0" encoding="utf-8"?> |
|||
<xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema"> |
|||
<!-- This file was generated by Fody. Manual changes to this file will be lost when your project is rebuilt. --> |
|||
<xs:element name="Weavers"> |
|||
<xs:complexType> |
|||
<xs:all> |
|||
<xs:element name="ConfigureAwait" minOccurs="0" maxOccurs="1"> |
|||
<xs:complexType> |
|||
<xs:attribute name="ContinueOnCapturedContext" type="xs:boolean" /> |
|||
</xs:complexType> |
|||
</xs:element> |
|||
</xs:all> |
|||
<xs:attribute name="VerifyAssembly" type="xs:boolean"> |
|||
<xs:annotation> |
|||
<xs:documentation>'true' to run assembly verification (PEVerify) on the target assembly after all weavers have been executed.</xs:documentation> |
|||
</xs:annotation> |
|||
</xs:attribute> |
|||
<xs:attribute name="VerifyIgnoreCodes" type="xs:string"> |
|||
<xs:annotation> |
|||
<xs:documentation>A comma-separated list of error codes that can be safely ignored in assembly verification.</xs:documentation> |
|||
</xs:annotation> |
|||
</xs:attribute> |
|||
<xs:attribute name="GenerateXsd" type="xs:boolean"> |
|||
<xs:annotation> |
|||
<xs:documentation>'false' to turn off automatic generation of the XML Schema file.</xs:documentation> |
|||
</xs:annotation> |
|||
</xs:attribute> |
|||
</xs:complexType> |
|||
</xs:element> |
|||
</xs:schema> |
|||
@ -0,0 +1,17 @@ |
|||
<Project Sdk="Microsoft.NET.Sdk"> |
|||
|
|||
<Import Project="..\..\..\configureawait.props" /> |
|||
<Import Project="..\..\..\common.props" /> |
|||
|
|||
<PropertyGroup> |
|||
<TargetFramework>netstandard2.0</TargetFramework> |
|||
<RootNamespace /> |
|||
</PropertyGroup> |
|||
|
|||
<ItemGroup> |
|||
<PackageReference Include="Confluent.Kafka" Version="1.5.0" /> |
|||
<ProjectReference Include="..\Volo.Abp.Json\Volo.Abp.Json.csproj" /> |
|||
<ProjectReference Include="..\Volo.Abp.Threading\Volo.Abp.Threading.csproj" /> |
|||
</ItemGroup> |
|||
|
|||
</Project> |
|||
@ -0,0 +1,31 @@ |
|||
using Microsoft.Extensions.DependencyInjection; |
|||
using Volo.Abp.Json; |
|||
using Volo.Abp.Modularity; |
|||
using Volo.Abp.Threading; |
|||
|
|||
namespace Volo.Abp.Kafka |
|||
{ |
|||
[DependsOn( |
|||
typeof(AbpJsonModule), |
|||
typeof(AbpThreadingModule) |
|||
)] |
|||
public class AbpKafkaModule : AbpModule |
|||
{ |
|||
public override void ConfigureServices(ServiceConfigurationContext context) |
|||
{ |
|||
var configuration = context.Services.GetConfiguration(); |
|||
Configure<AbpKafkaOptions>(configuration.GetSection("Kafka")); |
|||
} |
|||
|
|||
public override void OnApplicationShutdown(ApplicationShutdownContext context) |
|||
{ |
|||
context.ServiceProvider |
|||
.GetRequiredService<IConsumerPool>() |
|||
.Dispose(); |
|||
|
|||
context.ServiceProvider |
|||
.GetRequiredService<IProducerPool>() |
|||
.Dispose(); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,22 @@ |
|||
using System; |
|||
using Confluent.Kafka; |
|||
using Confluent.Kafka.Admin; |
|||
|
|||
namespace Volo.Abp.Kafka |
|||
{ |
|||
public class AbpKafkaOptions |
|||
{ |
|||
public KafkaConnections Connections { get; } |
|||
|
|||
public Action<ProducerConfig> ConfigureProducer { get; set; } |
|||
|
|||
public Action<ConsumerConfig> ConfigureConsumer { get; set; } |
|||
|
|||
public Action<TopicSpecification> ConfigureTopic { get; set; } |
|||
|
|||
public AbpKafkaOptions() |
|||
{ |
|||
Connections = new KafkaConnections(); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,109 @@ |
|||
using System; |
|||
using System.Collections.Concurrent; |
|||
using System.Collections.Generic; |
|||
using System.Diagnostics; |
|||
using System.Linq; |
|||
using Confluent.Kafka; |
|||
using Microsoft.Extensions.Logging; |
|||
using Microsoft.Extensions.Logging.Abstractions; |
|||
using Microsoft.Extensions.Options; |
|||
using Volo.Abp.DependencyInjection; |
|||
|
|||
namespace Volo.Abp.Kafka |
|||
{ |
|||
public class ConsumerPool : IConsumerPool, ISingletonDependency |
|||
{ |
|||
protected AbpKafkaOptions Options { get; } |
|||
|
|||
protected ConcurrentDictionary<string, IConsumer<string, byte[]>> Consumers { get; } |
|||
|
|||
protected TimeSpan TotalDisposeWaitDuration { get; set; } = TimeSpan.FromSeconds(10); |
|||
|
|||
public ILogger<ConsumerPool> Logger { get; set; } |
|||
|
|||
private bool _isDisposed; |
|||
|
|||
public ConsumerPool(IOptions<AbpKafkaOptions> options) |
|||
{ |
|||
Options = options.Value; |
|||
|
|||
Consumers = new ConcurrentDictionary<string, IConsumer<string, byte[]>>(); |
|||
Logger = new NullLogger<ConsumerPool>(); |
|||
} |
|||
|
|||
public virtual IConsumer<string, byte[]> Get(string groupId, string connectionName = null) |
|||
{ |
|||
connectionName ??= KafkaConnections.DefaultConnectionName; |
|||
|
|||
return Consumers.GetOrAdd( |
|||
connectionName, connection => |
|||
{ |
|||
var config = new ConsumerConfig(Options.Connections.GetOrDefault(connection)) |
|||
{ |
|||
GroupId = groupId, |
|||
EnableAutoCommit = false |
|||
}; |
|||
|
|||
Options.ConfigureConsumer?.Invoke(config); |
|||
|
|||
return new ConsumerBuilder<string, byte[]>(config).Build(); |
|||
} |
|||
); |
|||
} |
|||
|
|||
public void Dispose() |
|||
{ |
|||
if (_isDisposed) |
|||
{ |
|||
return; |
|||
} |
|||
|
|||
_isDisposed = true; |
|||
|
|||
if (!Consumers.Any()) |
|||
{ |
|||
Logger.LogDebug($"Disposed consumer pool with no consumers in the pool."); |
|||
return; |
|||
} |
|||
|
|||
var poolDisposeStopwatch = Stopwatch.StartNew(); |
|||
|
|||
Logger.LogInformation($"Disposing consumer pool ({Consumers.Count} consumers)."); |
|||
|
|||
var remainingWaitDuration = TotalDisposeWaitDuration; |
|||
|
|||
foreach (var consumer in Consumers.Values) |
|||
{ |
|||
var poolItemDisposeStopwatch = Stopwatch.StartNew(); |
|||
|
|||
try |
|||
{ |
|||
consumer.Close(); |
|||
consumer.Dispose(); |
|||
} |
|||
catch |
|||
{ |
|||
} |
|||
|
|||
poolItemDisposeStopwatch.Stop(); |
|||
|
|||
remainingWaitDuration = remainingWaitDuration > poolItemDisposeStopwatch.Elapsed |
|||
? remainingWaitDuration.Subtract(poolItemDisposeStopwatch.Elapsed) |
|||
: TimeSpan.Zero; |
|||
} |
|||
|
|||
poolDisposeStopwatch.Stop(); |
|||
|
|||
Logger.LogInformation( |
|||
$"Disposed Kafka Consumer Pool ({Consumers.Count} consumers in {poolDisposeStopwatch.Elapsed.TotalMilliseconds:0.00} ms)."); |
|||
|
|||
if (poolDisposeStopwatch.Elapsed.TotalSeconds > 5.0) |
|||
{ |
|||
Logger.LogWarning( |
|||
$"Disposing Kafka Consumer Pool got time greather than expected: {poolDisposeStopwatch.Elapsed.TotalMilliseconds:0.00} ms."); |
|||
} |
|||
|
|||
Consumers.Clear(); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,10 @@ |
|||
using System; |
|||
using Confluent.Kafka; |
|||
|
|||
namespace Volo.Abp.Kafka |
|||
{ |
|||
public interface IConsumerPool : IDisposable |
|||
{ |
|||
IConsumer<string, byte[]> Get(string groupId, string connectionName = null); |
|||
} |
|||
} |
|||
@ -0,0 +1,11 @@ |
|||
using System; |
|||
using System.Threading.Tasks; |
|||
using Confluent.Kafka; |
|||
|
|||
namespace Volo.Abp.Kafka |
|||
{ |
|||
public interface IKafkaMessageConsumer |
|||
{ |
|||
void OnMessageReceived(Func<Message<string, byte[]>, Task> callback); |
|||
} |
|||
} |
|||
@ -0,0 +1,19 @@ |
|||
namespace Volo.Abp.Kafka |
|||
{ |
|||
public interface IKafkaMessageConsumerFactory |
|||
{ |
|||
/// <summary>
|
|||
/// Creates a new <see cref="IKafkaMessageConsumer"/>.
|
|||
/// Avoid to create too many consumers since they are
|
|||
/// not disposed until end of the application.
|
|||
/// </summary>
|
|||
/// <param name="topicName"></param>
|
|||
/// <param name="groupId"></param>
|
|||
/// <param name="connectionName"></param>
|
|||
/// <returns></returns>
|
|||
IKafkaMessageConsumer Create( |
|||
string topicName, |
|||
string groupId, |
|||
string connectionName = null); |
|||
} |
|||
} |
|||
@ -0,0 +1,11 @@ |
|||
using System; |
|||
|
|||
namespace Volo.Abp.Kafka |
|||
{ |
|||
public interface IKafkaSerializer |
|||
{ |
|||
byte[] Serialize(object obj); |
|||
|
|||
object Deserialize(byte[] value, Type type); |
|||
} |
|||
} |
|||
@ -0,0 +1,10 @@ |
|||
using System; |
|||
using Confluent.Kafka; |
|||
|
|||
namespace Volo.Abp.Kafka |
|||
{ |
|||
public interface IProducerPool : IDisposable |
|||
{ |
|||
IProducer<string, byte[]> Get(string connectionName = null); |
|||
} |
|||
} |
|||
@ -0,0 +1,35 @@ |
|||
using System; |
|||
using System.Collections.Generic; |
|||
using Confluent.Kafka; |
|||
using JetBrains.Annotations; |
|||
|
|||
namespace Volo.Abp.Kafka |
|||
{ |
|||
[Serializable] |
|||
public class KafkaConnections : Dictionary<string, ClientConfig> |
|||
{ |
|||
public const string DefaultConnectionName = "Default"; |
|||
|
|||
[NotNull] |
|||
public ClientConfig Default |
|||
{ |
|||
get => this[DefaultConnectionName]; |
|||
set => this[DefaultConnectionName] = Check.NotNull(value, nameof(value)); |
|||
} |
|||
|
|||
public KafkaConnections() |
|||
{ |
|||
Default = new ClientConfig(); |
|||
} |
|||
|
|||
public ClientConfig GetOrDefault(string connectionName) |
|||
{ |
|||
if (TryGetValue(connectionName, out var connectionFactory)) |
|||
{ |
|||
return connectionFactory; |
|||
} |
|||
|
|||
return Default; |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,156 @@ |
|||
using System; |
|||
using System.Collections.Concurrent; |
|||
using System.Linq; |
|||
using System.Threading.Tasks; |
|||
using Confluent.Kafka; |
|||
using Confluent.Kafka.Admin; |
|||
using JetBrains.Annotations; |
|||
using Microsoft.Extensions.Logging; |
|||
using Microsoft.Extensions.Logging.Abstractions; |
|||
using Microsoft.Extensions.Options; |
|||
using Volo.Abp.DependencyInjection; |
|||
using Volo.Abp.ExceptionHandling; |
|||
using Volo.Abp.Threading; |
|||
|
|||
namespace Volo.Abp.Kafka |
|||
{ |
|||
public class KafkaMessageConsumer : IKafkaMessageConsumer, ITransientDependency, IDisposable |
|||
{ |
|||
public ILogger<KafkaMessageConsumer> Logger { get; set; } |
|||
|
|||
protected IConsumerPool ConsumerPool { get; } |
|||
|
|||
protected IExceptionNotifier ExceptionNotifier { get; } |
|||
|
|||
protected AbpKafkaOptions Options { get; } |
|||
|
|||
protected ConcurrentBag<Func<Message<string, byte[]>, Task>> Callbacks { get; } |
|||
|
|||
protected IConsumer<string, byte[]> Consumer { get; private set; } |
|||
|
|||
protected string ConnectionName { get; private set; } |
|||
|
|||
protected string GroupId { get; private set; } |
|||
|
|||
protected string TopicName { get; private set; } |
|||
|
|||
public KafkaMessageConsumer( |
|||
IConsumerPool consumerPool, |
|||
IExceptionNotifier exceptionNotifier, |
|||
IOptions<AbpKafkaOptions> options) |
|||
{ |
|||
ConsumerPool = consumerPool; |
|||
ExceptionNotifier = exceptionNotifier; |
|||
Options = options.Value; |
|||
Logger = NullLogger<KafkaMessageConsumer>.Instance; |
|||
|
|||
Callbacks = new ConcurrentBag<Func<Message<string, byte[]>, Task>>(); |
|||
} |
|||
|
|||
public virtual void Initialize( |
|||
[NotNull] string topicName, |
|||
[NotNull] string groupId, |
|||
string connectionName = null) |
|||
{ |
|||
Check.NotNull(topicName, nameof(topicName)); |
|||
Check.NotNull(groupId, nameof(groupId)); |
|||
TopicName = topicName; |
|||
ConnectionName = connectionName ?? KafkaConnections.DefaultConnectionName; |
|||
GroupId = groupId; |
|||
|
|||
AsyncHelper.RunSync(CreateTopicAsync); |
|||
Consume(); |
|||
} |
|||
|
|||
public virtual void OnMessageReceived(Func<Message<string, byte[]>, Task> callback) |
|||
{ |
|||
Callbacks.Add(callback); |
|||
} |
|||
|
|||
protected virtual async Task CreateTopicAsync() |
|||
{ |
|||
using (var adminClient = new AdminClientBuilder(Options.Connections.GetOrDefault(ConnectionName)).Build()) |
|||
{ |
|||
var topic = new TopicSpecification |
|||
{ |
|||
Name = TopicName, |
|||
NumPartitions = 1, |
|||
ReplicationFactor = 1 |
|||
}; |
|||
|
|||
Options.ConfigureTopic?.Invoke(topic); |
|||
|
|||
try |
|||
{ |
|||
await adminClient.CreateTopicsAsync(new[] {topic}); |
|||
} |
|||
catch (CreateTopicsException e) |
|||
{ |
|||
if (!e.Error.Reason.Contains($"Topic '{TopicName}' already exists")) |
|||
{ |
|||
throw; |
|||
} |
|||
} |
|||
} |
|||
} |
|||
|
|||
protected virtual void Consume() |
|||
{ |
|||
Consumer = ConsumerPool.Get(GroupId, ConnectionName); |
|||
|
|||
Task.Factory.StartNew(async () => |
|||
{ |
|||
Consumer.Subscribe(TopicName); |
|||
|
|||
while (true) |
|||
{ |
|||
try |
|||
{ |
|||
var consumeResult = Consumer.Consume(); |
|||
|
|||
if (consumeResult.IsPartitionEOF) |
|||
{ |
|||
continue; |
|||
} |
|||
|
|||
await HandleIncomingMessage(consumeResult); |
|||
} |
|||
catch (ConsumeException ex) |
|||
{ |
|||
Logger.LogException(ex, LogLevel.Warning); |
|||
AsyncHelper.RunSync(() => ExceptionNotifier.NotifyAsync(ex, logLevel: LogLevel.Warning)); |
|||
} |
|||
} |
|||
}); |
|||
} |
|||
|
|||
protected virtual async Task HandleIncomingMessage(ConsumeResult<string, byte[]> consumeResult) |
|||
{ |
|||
try |
|||
{ |
|||
foreach (var callback in Callbacks) |
|||
{ |
|||
await callback(consumeResult.Message); |
|||
} |
|||
|
|||
Consumer.Commit(consumeResult); |
|||
} |
|||
catch (Exception ex) |
|||
{ |
|||
Logger.LogException(ex); |
|||
await ExceptionNotifier.NotifyAsync(ex); |
|||
} |
|||
} |
|||
|
|||
public virtual void Dispose() |
|||
{ |
|||
if (Consumer == null) |
|||
{ |
|||
return; |
|||
} |
|||
|
|||
Consumer.Close(); |
|||
Consumer.Dispose(); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,32 @@ |
|||
using System; |
|||
using System.Threading.Tasks; |
|||
using Microsoft.Extensions.DependencyInjection; |
|||
using Volo.Abp.DependencyInjection; |
|||
|
|||
namespace Volo.Abp.Kafka |
|||
{ |
|||
public class KafkaMessageConsumerFactory : IKafkaMessageConsumerFactory, ISingletonDependency, IDisposable |
|||
{ |
|||
protected IServiceScope ServiceScope { get; } |
|||
|
|||
public KafkaMessageConsumerFactory(IServiceScopeFactory serviceScopeFactory) |
|||
{ |
|||
ServiceScope = serviceScopeFactory.CreateScope(); |
|||
} |
|||
|
|||
public IKafkaMessageConsumer Create( |
|||
string topicName, |
|||
string groupId, |
|||
string connectionName = null) |
|||
{ |
|||
var consumer = ServiceScope.ServiceProvider.GetRequiredService<KafkaMessageConsumer>(); |
|||
consumer.Initialize(topicName, groupId, connectionName); |
|||
return consumer; |
|||
} |
|||
|
|||
public void Dispose() |
|||
{ |
|||
ServiceScope?.Dispose(); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,102 @@ |
|||
using System; |
|||
using System.Collections.Concurrent; |
|||
using System.Diagnostics; |
|||
using System.Linq; |
|||
using Confluent.Kafka; |
|||
using Microsoft.Extensions.Logging; |
|||
using Microsoft.Extensions.Logging.Abstractions; |
|||
using Microsoft.Extensions.Options; |
|||
using Volo.Abp.DependencyInjection; |
|||
|
|||
namespace Volo.Abp.Kafka |
|||
{ |
|||
public class ProducerPool : IProducerPool, ISingletonDependency |
|||
{ |
|||
protected AbpKafkaOptions Options { get; } |
|||
|
|||
protected ConcurrentDictionary<string, IProducer<string, byte[]>> Producers { get; } |
|||
|
|||
protected TimeSpan TotalDisposeWaitDuration { get; set; } = TimeSpan.FromSeconds(10); |
|||
|
|||
public ILogger<ProducerPool> Logger { get; set; } |
|||
|
|||
private bool _isDisposed; |
|||
|
|||
public ProducerPool(IOptions<AbpKafkaOptions> options) |
|||
{ |
|||
Options = options.Value; |
|||
|
|||
Producers = new ConcurrentDictionary<string, IProducer<string, byte[]>>(); |
|||
Logger = new NullLogger<ProducerPool>(); |
|||
} |
|||
|
|||
public virtual IProducer<string, byte[]> Get(string connectionName = null) |
|||
{ |
|||
connectionName ??= KafkaConnections.DefaultConnectionName; |
|||
|
|||
return Producers.GetOrAdd( |
|||
connectionName, connection => |
|||
{ |
|||
var config = Options.Connections.GetOrDefault(connection); |
|||
|
|||
Options.ConfigureProducer?.Invoke(new ProducerConfig(config)); |
|||
|
|||
return new ProducerBuilder<string, byte[]>(config).Build(); |
|||
}); |
|||
} |
|||
|
|||
public void Dispose() |
|||
{ |
|||
if (_isDisposed) |
|||
{ |
|||
return; |
|||
} |
|||
|
|||
_isDisposed = true; |
|||
|
|||
if (!Producers.Any()) |
|||
{ |
|||
Logger.LogDebug($"Disposed producer pool with no producers in the pool."); |
|||
return; |
|||
} |
|||
|
|||
var poolDisposeStopwatch = Stopwatch.StartNew(); |
|||
|
|||
Logger.LogInformation($"Disposing producer pool ({Producers.Count} producers)."); |
|||
|
|||
var remainingWaitDuration = TotalDisposeWaitDuration; |
|||
|
|||
foreach (var producer in Producers.Values) |
|||
{ |
|||
var poolItemDisposeStopwatch = Stopwatch.StartNew(); |
|||
|
|||
try |
|||
{ |
|||
producer.Dispose(); |
|||
} |
|||
catch |
|||
{ |
|||
} |
|||
|
|||
poolItemDisposeStopwatch.Stop(); |
|||
|
|||
remainingWaitDuration = remainingWaitDuration > poolItemDisposeStopwatch.Elapsed |
|||
? remainingWaitDuration.Subtract(poolItemDisposeStopwatch.Elapsed) |
|||
: TimeSpan.Zero; |
|||
} |
|||
|
|||
poolDisposeStopwatch.Stop(); |
|||
|
|||
Logger.LogInformation( |
|||
$"Disposed Kafka Producer Pool ({Producers.Count} producers in {poolDisposeStopwatch.Elapsed.TotalMilliseconds:0.00} ms)."); |
|||
|
|||
if (poolDisposeStopwatch.Elapsed.TotalSeconds > 5.0) |
|||
{ |
|||
Logger.LogWarning( |
|||
$"Disposing Kafka Producer Pool got time greather than expected: {poolDisposeStopwatch.Elapsed.TotalMilliseconds:0.00} ms."); |
|||
} |
|||
|
|||
Producers.Clear(); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,27 @@ |
|||
using System; |
|||
using System.Text; |
|||
using Volo.Abp.DependencyInjection; |
|||
using Volo.Abp.Json; |
|||
|
|||
namespace Volo.Abp.Kafka |
|||
{ |
|||
public class Utf8JsonKafkaSerializer : IKafkaSerializer, ITransientDependency |
|||
{ |
|||
private readonly IJsonSerializer _jsonSerializer; |
|||
|
|||
public Utf8JsonKafkaSerializer(IJsonSerializer jsonSerializer) |
|||
{ |
|||
_jsonSerializer = jsonSerializer; |
|||
} |
|||
|
|||
public byte[] Serialize(object obj) |
|||
{ |
|||
return Encoding.UTF8.GetBytes(_jsonSerializer.Serialize(obj)); |
|||
} |
|||
|
|||
public object Deserialize(byte[] value, Type type) |
|||
{ |
|||
return _jsonSerializer.Deserialize(type, Encoding.UTF8.GetString(value)); |
|||
} |
|||
} |
|||
} |
|||
Loading…
Reference in new issue