-
Notifications
You must be signed in to change notification settings - Fork 322
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Browse files
Browse the repository at this point in the history
* feat:add EasyCaching.Bus.ConfluentKafka module 1.use kafka to notify cache to change 2.testing seeing sample EasyCaching.Demo.Providers ValuesBusController * feat: 1.Add ZooKeeperBus Module, 2.kafakBus options write AutoOffsetReset value * feat:zookeeperBus * feat: 1.zk bus * fix: 1.add zk config command,zk init time compare 2.add testBus * feat： 1.Eliminate careless writing and code formatting * fix: 1.edit build/releasenotes.props and build/version.props add zookeeper/kafka bus module notes and version 2.EasyCaching.Demo.Providers/Startup.cs to original file data
- Loading branch information
1 parent
c32dae1
commit 14bbda4
Showing
19 changed files
with
1,203 additions
and
134 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
39 changes: 39 additions & 0 deletions
39
bus/EasyCaching.Bus.ConfluentKafka/Configurations/ConfluentKafkaBusOptions.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,39 @@ | ||
| using Confluent.Kafka; | ||
|
|
||
| namespace EasyCaching.Bus.ConfluentKafka | ||
| { | ||
| /// <summary> | ||
| /// kafka bus options | ||
| /// </summary> | ||
| public class ConfluentKafkaBusOptions | ||
| { | ||
| /// <summary> | ||
| /// kafka address(BootstrapServers must) | ||
| /// </summary> | ||
| public string BootstrapServers { get; set; } | ||
|
|
||
|
|
||
| /// <summary> | ||
| /// kafka bus producer options. | ||
| /// </summary> | ||
| public ProducerConfig ProducerConfig { get; set; } | ||
|
|
||
| /// <summary> | ||
| /// kafka bus consumer options.(if GroupId value below is empty,then ConsumerConfig.GroupId must ) | ||
| /// </summary> | ||
| public ConsumerConfig ConsumerConfig { get; set; } | ||
|
|
||
| /// <summary> | ||
| /// kafka bus consumer options with consumer groupId | ||
| /// (if ConsumerConfig below has give GroupId value , this options can ignore) | ||
| /// import:if application is cluster,you should set this different value in application,this will make consumer can consumerdata | ||
| /// </summary> | ||
| public string GroupId { get; set; } | ||
|
|
||
| /// <summary> | ||
| /// kafka bus consumer consume count | ||
| /// </summary> | ||
| public int ConsumerCount { get; set; } = 1; | ||
| } | ||
|
|
||
| } |
45 changes: 45 additions & 0 deletions
45
bus/EasyCaching.Bus.ConfluentKafka/Configurations/ConfluentKafkaOptionsExtension.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,45 @@ | ||
| namespace EasyCaching.Bus.ConfluentKafka | ||
| { | ||
| using System; | ||
| using EasyCaching.Core.Bus; | ||
| using EasyCaching.Core.Configurations; | ||
| using EasyCaching.Core.Serialization; | ||
| using Microsoft.Extensions.DependencyInjection; | ||
| using Microsoft.Extensions.DependencyInjection.Extensions; | ||
|
|
||
| /// <summary> | ||
| /// Kafka options extension. | ||
| /// </summary> | ||
| internal sealed class ConfluentKafkaOptionsExtension : IEasyCachingOptionsExtension | ||
| { | ||
|
|
||
| private readonly Action<ConfluentKafkaBusOptions> _confluentKafkaBusOptions; | ||
|
|
||
| public ConfluentKafkaOptionsExtension(Action<ConfluentKafkaBusOptions> confluentKafkaBusOptions) | ||
| { | ||
| this._confluentKafkaBusOptions = confluentKafkaBusOptions; | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Adds the services. | ||
| /// </summary> | ||
| /// <param name="services">Services.</param> | ||
| public void AddServices(IServiceCollection services) | ||
| { | ||
| services.AddOptions(); | ||
|
|
||
| services.TryAddSingleton<IEasyCachingSerializer, DefaultBinaryFormatterSerializer>(); | ||
|
|
||
| services.AddOptions<ConfluentKafkaBusOptions>() | ||
| .Configure(_confluentKafkaBusOptions); | ||
|
|
||
|
|
||
| //var options = services.BuildServiceProvider() | ||
| // .GetRequiredService<IOptions<ConfluentKafkaBusOptions>>() | ||
| // .Value; | ||
|
|
||
| services.AddSingleton<IEasyCachingBus, DefaultConfluentKafkaBus>(); | ||
|
|
||
| } | ||
| } | ||
| } |
81 changes: 81 additions & 0 deletions
81
bus/EasyCaching.Bus.ConfluentKafka/Configurations/EasyCachingOptionsExtensions.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,81 @@ | ||
| namespace Microsoft.Extensions.DependencyInjection | ||
| { | ||
| using Confluent.Kafka; | ||
| using EasyCaching.Bus.ConfluentKafka; | ||
| using EasyCaching.Core; | ||
| using EasyCaching.Core.Configurations; | ||
| using Microsoft.Extensions.Configuration; | ||
| using System; | ||
|
|
||
| /// <summary> | ||
| /// EasyCaching options extensions. | ||
| /// </summary> | ||
| public static class EasyCachingOptionsExtensions | ||
| { | ||
| /// <summary> | ||
| /// Withs the ConfluentKafka bus (specify the config via hard code). | ||
| /// </summary> | ||
| /// <param name="options"></param> | ||
| /// <param name="configure"></param> | ||
| /// <returns></returns> | ||
| public static EasyCachingOptions WithConfluentKafkaBus( | ||
| this EasyCachingOptions options | ||
| , Action<ConfluentKafkaBusOptions> configure | ||
| ) | ||
| { | ||
| ArgumentCheck.NotNull(configure, nameof(configure)); | ||
| //option convert | ||
| ConfluentKafkaBusOptions kafkaOptions = new ConfluentKafkaBusOptions(); | ||
| configure.Invoke(kafkaOptions); | ||
| void kafkaBusConfigure(ConfluentKafkaBusOptions x) | ||
| { | ||
| x.BootstrapServers = kafkaOptions.BootstrapServers; | ||
| x.ProducerConfig = kafkaOptions.ProducerConfig ?? new ProducerConfig(); | ||
| x.ConsumerConfig = kafkaOptions.ConsumerConfig ?? new ConsumerConfig(); | ||
| //address | ||
| x.ProducerConfig.BootstrapServers = x.ProducerConfig.BootstrapServers ?? kafkaOptions.BootstrapServers; | ||
| x.ConsumerConfig.BootstrapServers = x.ConsumerConfig.BootstrapServers ?? kafkaOptions.BootstrapServers; | ||
| //consumer groupId | ||
| x.ConsumerConfig.GroupId = x.ConsumerConfig.GroupId ?? kafkaOptions.GroupId; | ||
| x.ConsumerConfig.AutoOffsetReset = kafkaOptions.ConsumerConfig.AutoOffsetReset ?? AutoOffsetReset.Latest; | ||
| } | ||
|
|
||
| options.RegisterExtension(new ConfluentKafkaOptionsExtension(kafkaBusConfigure)); | ||
| return options; | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Withs the ConfluentKafka bus (read config from configuration file). | ||
| /// </summary> | ||
| /// <param name="options"></param> | ||
| /// <param name="configuration"></param> | ||
| /// <param name="sectionName">The section name in the configuration file.</param> | ||
| /// <returns></returns> | ||
| public static EasyCachingOptions WithConfluentKafkaBus( | ||
| this EasyCachingOptions options | ||
| , IConfiguration configuration | ||
| , string sectionName = EasyCachingConstValue.KafkaBusSection | ||
| ) | ||
| { | ||
| var dbConfig = configuration.GetSection(sectionName); | ||
| var kafkaOptions = new ConfluentKafkaBusOptions(); | ||
| dbConfig.Bind(kafkaOptions); | ||
|
|
||
| void configure(ConfluentKafkaBusOptions x) | ||
| { | ||
| x.BootstrapServers = kafkaOptions.BootstrapServers; | ||
| x.ProducerConfig = kafkaOptions.ProducerConfig ?? new ProducerConfig(); | ||
| x.ConsumerConfig = kafkaOptions.ConsumerConfig ?? new ConsumerConfig(); | ||
| //address | ||
| x.ProducerConfig.BootstrapServers = x.ProducerConfig.BootstrapServers ?? kafkaOptions.BootstrapServers; | ||
| x.ConsumerConfig.BootstrapServers = x.ConsumerConfig.BootstrapServers ?? kafkaOptions.BootstrapServers; | ||
| //consumer groupId | ||
| x.ConsumerConfig.GroupId = x.ConsumerConfig.GroupId ?? kafkaOptions.GroupId; | ||
| x.ConsumerConfig.AutoOffsetReset = kafkaOptions.ConsumerConfig.AutoOffsetReset ?? AutoOffsetReset.Latest; | ||
| } | ||
|
|
||
| options.RegisterExtension(new ConfluentKafkaOptionsExtension(configure)); | ||
| return options; | ||
| } | ||
| } | ||
| } |
149 changes: 149 additions & 0 deletions
149
bus/EasyCaching.Bus.ConfluentKafka/DefaultConfluentKafkaBus.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,149 @@ | ||
| namespace EasyCaching.Bus.ConfluentKafka | ||
| { | ||
| using System; | ||
| using System.Threading; | ||
| using System.Threading.Tasks; | ||
| using Confluent.Kafka; | ||
| using EasyCaching.Core; | ||
| using EasyCaching.Core.Bus; | ||
| using EasyCaching.Core.Serialization; | ||
| using Microsoft.Extensions.Logging; | ||
| using Microsoft.Extensions.Logging.Abstractions; | ||
| using Microsoft.Extensions.Options; | ||
|
|
||
| public class DefaultConfluentKafkaBus : EasyCachingAbstractBus | ||
| { | ||
|
|
||
|
|
||
| /// <summary> | ||
| /// The kafka Bus options. | ||
| /// </summary> | ||
| private readonly ConfluentKafkaBusOptions _kafkaBusOptions; | ||
|
|
||
| /// <summary> | ||
| /// The serializer. | ||
| /// </summary> | ||
| private readonly IEasyCachingSerializer _serializer; | ||
|
|
||
| /// <summary> | ||
| /// kafka producer object | ||
| /// </summary> | ||
|
|
||
| private readonly IProducer<Null, byte[]> _producer; | ||
|
|
||
|
|
||
| /// <summary> | ||
| /// log | ||
| /// </summary> | ||
|
|
||
| private readonly ILogger _logger = NullLogger<DefaultConfluentKafkaBus>.Instance; | ||
|
|
||
|
|
||
| /// <summary> | ||
| /// Initializes a new instance of the <see cref="T:EasyCaching.Bus.ConfluentKafka.DefaultConfluentKafkaBus"/> class. | ||
| /// </summary> | ||
| /// <param name="kafkaBusOptions"></param> | ||
| /// <param name="serializer"></param> | ||
| public DefaultConfluentKafkaBus( | ||
| IOptionsMonitor<ConfluentKafkaBusOptions> kafkaBusOptions | ||
| , IEasyCachingSerializer serializer) | ||
| { | ||
| this.BusName = "easycachingbus"; | ||
| this._kafkaBusOptions = kafkaBusOptions.CurrentValue; | ||
|
|
||
| this._producer = new ProducerBuilder<Null, byte[]>(this._kafkaBusOptions.ProducerConfig).Build(); | ||
|
|
||
| this._serializer = serializer; | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Publish the specified topic and message. | ||
| /// </summary> | ||
| /// <param name="topic">Topic.</param> | ||
| /// <param name="message">Message.</param> | ||
| public override void BasePublish(string topic, EasyCachingMessage message) | ||
| { | ||
| var msg = _serializer.Serialize(message); | ||
|
|
||
| _producer.Produce(topic, new Message<Null, byte[]> { Value = msg }); | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Publishs the async. | ||
| /// </summary> | ||
| /// <returns>The async.</returns> | ||
| /// <param name="topic">Topic.</param> | ||
| /// <param name="message">Message.</param> | ||
| /// <param name="cancellationToken">Cancellation token.</param> | ||
| public override async Task BasePublishAsync(string topic, EasyCachingMessage message, CancellationToken cancellationToken = default(CancellationToken)) | ||
| { | ||
| var msg = _serializer.Serialize(message); | ||
|
|
||
| await _producer.ProduceAsync(topic, new Message<Null, byte[]> { Value = msg }); | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Subscribe the specified topic and action. | ||
| /// </summary> | ||
| /// <param name="topic">Topic.</param> | ||
| /// <param name="action">Action.</param> | ||
| public override void BaseSubscribe(string topic, Action<EasyCachingMessage> action) | ||
| { | ||
| Task.Factory.StartNew(() => | ||
| { | ||
| for (int i = 0; i < this._kafkaBusOptions.ConsumerCount; i++) | ||
| { | ||
| using (var consumer = new ConsumerBuilder<Null, byte[]>(this._kafkaBusOptions.ConsumerConfig).Build()) | ||
| { | ||
| consumer.Subscribe(topic); | ||
| try | ||
| { | ||
| while (true) | ||
| { | ||
| try | ||
| { | ||
| var cr = consumer.Consume(); | ||
| if (cr.IsPartitionEOF | ||
| || cr.Message == null | ||
| || cr.Message.Value.Length == 0) | ||
| { | ||
| continue; | ||
| } | ||
| OnMessage(cr.Message.Value); | ||
| } | ||
| catch (ConsumeException ex) | ||
| { | ||
| _logger.LogError(ex, "Consumer {0} error of reason {1}.", topic, ex.Error.Reason); | ||
| } | ||
| catch (OperationCanceledException) | ||
| { | ||
| throw; | ||
| } | ||
| catch (Exception ex) | ||
| { | ||
| _logger.LogError(ex, "Consumer {0} error.", topic); | ||
| } | ||
| } | ||
| } | ||
| catch (OperationCanceledException ex) | ||
| { | ||
| _logger.LogWarning(ex, "Consumer {0} cancel.", topic); | ||
| consumer.Close(); | ||
| } | ||
| } | ||
| } | ||
| }, TaskCreationOptions.LongRunning); | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Ons the message. | ||
| /// </summary> | ||
| /// <param name="body">Body.</param> | ||
| private void OnMessage(byte[] body) | ||
| { | ||
| var message = _serializer.Deserialize<EasyCachingMessage>(body); | ||
|
|
||
| BaseOnMessage(message); | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.