admin/JiShe.CollectBus
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

规范代码

Browse Source
This commit is contained in:
cli 2025-04-21 10:17:40 +08:00
parent 01aeaebb0a
commit 02f2a2cafc
29 changed files with 604 additions and 689 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
modules/JiShe.CollectBus.IoTDB/CollectBusIoTDBModule.cs
View File

@ -1,33 +1,22 @@
using JiShe.CollectBus.IoTDB.Context; using JiShe.CollectBus.IoTDB.Context;
using JiShe.CollectBus.IoTDB.Interface;
using JiShe.CollectBus.IoTDB.Options; using JiShe.CollectBus.IoTDB.Options;
using JiShe.CollectBus.IoTDB.Provider;
using Microsoft.Extensions.Configuration; using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.DependencyInjection;
using Volo.Abp.Modularity; using Volo.Abp.Modularity;
namespace JiShe.CollectBus.IoTDB namespace JiShe.CollectBus.IoTDB;
/// <summary>
/// CollectBusIoTDBModule
/// </summary>
public class CollectBusIoTDbModule : AbpModule
{ {
public class CollectBusIoTDBModule : AbpModule public override void ConfigureServices(ServiceConfigurationContext context)
{ {
public override void ConfigureServices(ServiceConfigurationContext context) var configuration = context.Services.GetConfiguration();
{ Configure<IoTDbOptions>(options => { configuration.GetSection(nameof(IoTDbOptions)).Bind(options); });
var configuration = context.Services.GetConfiguration(); // 注册上下文为Scoped
Configure<IoTDBOptions>(options => context.Services.AddScoped<IoTDbRuntimeContext>();
{
configuration.GetSection(nameof(IoTDBOptions)).Bind(options);
});
// 注册上下文为Scoped
context.Services.AddScoped<IoTDBRuntimeContext>();
// 注册Session工厂
context.Services.AddSingleton<IIoTDBSessionFactory, IoTDBSessionFactory>();
// 注册Provider
context.Services.AddScoped<IIoTDBProvider, IoTDBProvider>();
}
} }
} }

4
modules/JiShe.CollectBus.IoTDB/Context/IoTDBRuntimeContext.cs
View File

@ -6,11 +6,11 @@ namespace JiShe.CollectBus.IoTDB.Context
/// <summary> /// <summary>
/// IoTDB SessionPool 运行时上下文 /// IoTDB SessionPool 运行时上下文
/// </summary> /// </summary>
public class IoTDBRuntimeContext public class IoTDbRuntimeContext
{ {
private readonly bool _defaultValue; private readonly bool _defaultValue;
public IoTDBRuntimeContext(IOptions<IoTDBOptions> options) public IoTDbRuntimeContext(IOptions<IoTDbOptions> options)
{ {
_defaultValue = options.Value.UseTableSessionPoolByDefault; _defaultValue = options.Value.UseTableSessionPoolByDefault;
UseTableSessionPool = _defaultValue; UseTableSessionPool = _defaultValue;

2
modules/JiShe.CollectBus.IoTDB/Interface/IIoTDBProvider.cs
View File

@ -7,7 +7,7 @@ namespace JiShe.CollectBus.IoTDB.Interface
/// <summary> /// <summary>
/// IoTDB数据源,数据库能同时存多个时序模型，但数据是完全隔离的，不能跨时序模型查询，通过连接字符串配置 /// IoTDB数据源,数据库能同时存多个时序模型，但数据是完全隔离的，不能跨时序模型查询，通过连接字符串配置
/// </summary> /// </summary>
public interface IIoTDBProvider public interface IIoTDbProvider
{ {
///// <summary> ///// <summary>
///// 切换 SessionPool ///// 切换 SessionPool

4
modules/JiShe.CollectBus.IoTDB/Interface/IIoTDBSessionFactory.cs
View File

@ -3,8 +3,8 @@
/// <summary> /// <summary>
/// Session 工厂接口 /// Session 工厂接口
/// </summary> /// </summary>
public interface IIoTDBSessionFactory:IDisposable public interface IIoTDbSessionFactory:IDisposable
{ {
IIoTDBSessionPool GetSessionPool(bool useTableSession); IIoTDbSessionPool GetSessionPool(bool useTableSession);
} }
} }

2
modules/JiShe.CollectBus.IoTDB/Interface/IIoTDBSessionPool.cs
View File

@ -5,7 +5,7 @@ namespace JiShe.CollectBus.IoTDB.Interface
/// <summary> /// <summary>
/// Session 连接池 /// Session 连接池
/// </summary> /// </summary>
public interface IIoTDBSessionPool : IDisposable public interface IIoTDbSessionPool : IDisposable
{ {
/// <summary> /// <summary>
/// 打开连接池 /// 打开连接池

2
modules/JiShe.CollectBus.IoTDB/Options/IoTDBOptions.cs
View File

@ -3,7 +3,7 @@
/// <summary> /// <summary>
/// IOTDB配置 /// IOTDB配置
/// </summary> /// </summary>
public class IoTDBOptions public class IoTDbOptions
{ {
/// <summary> /// <summary>
/// 数据库名称，表模型才有，树模型为空 /// 数据库名称，表模型才有，树模型为空

29
modules/JiShe.CollectBus.IoTDB/Provider/IoTDBProvider.cs
View File

@ -9,26 +9,33 @@ using JiShe.CollectBus.IoTDB.Context;
using JiShe.CollectBus.IoTDB.Interface; using JiShe.CollectBus.IoTDB.Interface;
using JiShe.CollectBus.IoTDB.Options; using JiShe.CollectBus.IoTDB.Options;
using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging;
using Volo.Abp.DependencyInjection;
namespace JiShe.CollectBus.IoTDB.Provider namespace JiShe.CollectBus.IoTDB.Provider
{ {
/// <summary> /// <summary>
/// IoTDB数据源 /// IoTDB数据源
/// </summary> /// </summary>
public class IoTDBProvider : IIoTDBProvider public class IoTDbProvider : IIoTDbProvider, IScopedDependency
{ {
private static readonly ConcurrentDictionary<Type, DeviceMetadata> _metadataCache = new(); private static readonly ConcurrentDictionary<Type, DeviceMetadata> MetadataCache = new();
private readonly ILogger<IoTDBProvider> _logger; private readonly ILogger<IoTDbProvider> _logger;
private readonly IIoTDBSessionFactory _sessionFactory; private readonly IIoTDbSessionFactory _sessionFactory;
private readonly IoTDBRuntimeContext _runtimeContext; private readonly IoTDbRuntimeContext _runtimeContext;
private IIoTDBSessionPool CurrentSession => private IIoTDbSessionPool CurrentSession =>
_sessionFactory.GetSessionPool(_runtimeContext.UseTableSessionPool); _sessionFactory.GetSessionPool(_runtimeContext.UseTableSessionPool);
public IoTDBProvider( /// <summary>
ILogger<IoTDBProvider> logger, /// IoTDbProvider
IIoTDBSessionFactory sessionFactory, /// </summary>
IoTDBRuntimeContext runtimeContext) /// <param name="logger"></param>
/// <param name="sessionFactory"></param>
/// <param name="runtimeContext"></param>
public IoTDbProvider(
ILogger<IoTDbProvider> logger,
IIoTDbSessionFactory sessionFactory,
IoTDbRuntimeContext runtimeContext)
{ {
_logger = logger; _logger = logger;
_sessionFactory = sessionFactory; _sessionFactory = sessionFactory;
@ -396,7 +403,7 @@ namespace JiShe.CollectBus.IoTDB.Provider
var columns = CollectColumnMetadata(typeof(T)); var columns = CollectColumnMetadata(typeof(T));
var metadata = BuildDeviceMetadata(columns); var metadata = BuildDeviceMetadata(columns);
return _metadataCache.AddOrUpdate( return MetadataCache.AddOrUpdate(
typeof(T), typeof(T),
addValueFactory: t => metadata, // 如果键不存在，用此值添加 addValueFactory: t => metadata, // 如果键不存在，用此值添加
updateValueFactory: (t, existingValue) => updateValueFactory: (t, existingValue) =>

19
modules/JiShe.CollectBus.IoTDB/Provider/IoTDBSessionFactory.cs
View File

@ -2,6 +2,7 @@
using JiShe.CollectBus.IoTDB.Interface; using JiShe.CollectBus.IoTDB.Interface;
using JiShe.CollectBus.IoTDB.Options; using JiShe.CollectBus.IoTDB.Options;
using Microsoft.Extensions.Options; using Microsoft.Extensions.Options;
using Volo.Abp.DependencyInjection;
namespace JiShe.CollectBus.IoTDB.Provider namespace JiShe.CollectBus.IoTDB.Provider
{ {
@ -9,25 +10,29 @@ namespace JiShe.CollectBus.IoTDB.Provider
/// <summary> /// <summary>
/// 实现带缓存的Session工厂 /// 实现带缓存的Session工厂
/// </summary> /// </summary>
public class IoTDBSessionFactory : IIoTDBSessionFactory public class IoTDbSessionFactory : IIoTDbSessionFactory, ISingletonDependency
{ {
private readonly IoTDBOptions _options; private readonly IoTDbOptions _options;
private readonly ConcurrentDictionary<bool, IIoTDBSessionPool> _pools = new(); private readonly ConcurrentDictionary<bool, IIoTDbSessionPool> _pools = new();
private bool _disposed; private bool _disposed;
public IoTDBSessionFactory(IOptions<IoTDBOptions> options) /// <summary>
/// IoTDbSessionFactory
/// </summary>
/// <param name="options"></param>
public IoTDbSessionFactory(IOptions<IoTDbOptions> options)
{ {
_options = options.Value; _options = options.Value;
} }
public IIoTDBSessionPool GetSessionPool(bool useTableSession) public IIoTDbSessionPool GetSessionPool(bool useTableSession)
{ {
if (_disposed) throw new ObjectDisposedException(nameof(IoTDBSessionFactory)); if (_disposed) throw new ObjectDisposedException(nameof(IoTDbSessionFactory));
return _pools.GetOrAdd(useTableSession, key => return _pools.GetOrAdd(useTableSession, key =>
{ {
var pool = key var pool = key
? (IIoTDBSessionPool)new TableSessionPoolAdapter(_options) ? (IIoTDbSessionPool)new TableSessionPoolAdapter(_options)
: new SessionPoolAdapter(_options); : new SessionPoolAdapter(_options);
pool.OpenAsync().ConfigureAwait(false).GetAwaiter().GetResult(); ; pool.OpenAsync().ConfigureAwait(false).GetAwaiter().GetResult(); ;

10
modules/JiShe.CollectBus.IoTDB/Provider/SessionPoolAdapter.cs
View File

@ -9,12 +9,16 @@ namespace JiShe.CollectBus.IoTDB.Provider
/// <summary> /// <summary>
/// 树模型连接池 /// 树模型连接池
/// </summary> /// </summary>
public class SessionPoolAdapter : IIoTDBSessionPool public class SessionPoolAdapter : IIoTDbSessionPool
{ {
private readonly SessionPool _sessionPool; private readonly SessionPool _sessionPool;
private readonly IoTDBOptions _options; private readonly IoTDbOptions _options;
public SessionPoolAdapter(IoTDBOptions options) /// <summary>
/// SessionPoolAdapter
/// </summary>
/// <param name="options"></param>
public SessionPoolAdapter(IoTDbOptions options)
{ {
_options = options; _options = options;
_sessionPool = new SessionPool.Builder() _sessionPool = new SessionPool.Builder()

10
modules/JiShe.CollectBus.IoTDB/Provider/TableSessionPoolAdapter.cs
View File

@ -9,12 +9,16 @@ namespace JiShe.CollectBus.IoTDB.Provider
/// <summary> /// <summary>
/// 表模型Session连接池 /// 表模型Session连接池
/// </summary> /// </summary>
public class TableSessionPoolAdapter : IIoTDBSessionPool public class TableSessionPoolAdapter : IIoTDbSessionPool
{ {
private readonly TableSessionPool _sessionPool; private readonly TableSessionPool _sessionPool;
private readonly IoTDBOptions _options; private readonly IoTDbOptions _options;
public TableSessionPoolAdapter(IoTDBOptions options) /// <summary>
/// TableSessionPoolAdapter
/// </summary>
/// <param name="options"></param>
public TableSessionPoolAdapter(IoTDbOptions options)
{ {
_options = options; _options = options;
_sessionPool = new TableSessionPool.Builder() _sessionPool = new TableSessionPool.Builder()

354
modules/JiShe.CollectBus.Kafka/AdminClient/AdminClientService.cs
View File

@ -1,204 +1,190 @@
using Confluent.Kafka; using Confluent.Kafka;
using Microsoft.Extensions.Configuration;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Confluent.Kafka.Admin; using Confluent.Kafka.Admin;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging;
using Volo.Abp.DependencyInjection; using Volo.Abp.DependencyInjection;
namespace JiShe.CollectBus.Kafka.AdminClient namespace JiShe.CollectBus.Kafka.AdminClient;
public class AdminClientService : IAdminClientService, IDisposable, ISingletonDependency
{ {
public class AdminClientService : IAdminClientService, IDisposable,ISingletonDependency private readonly ILogger<AdminClientService> _logger;
/// <summary>
/// Initializes a new instance of the <see cref="AdminClientService" /> class.
/// </summary>
/// <param name="configuration"></param>
/// <param name="logger"></param>
public AdminClientService(IConfiguration configuration, ILogger<AdminClientService> logger)
{ {
_logger = logger;
Instance = GetInstance(configuration);
}
private readonly ILogger<AdminClientService> _logger; /// <summary>
/// Gets or sets the instance.
/// </summary>
/// <value>
/// The instance.
/// </value>
public IAdminClient Instance { get; set; }
/// <summary> /// <summary>
/// Initializes a new instance of the <see cref="AdminClientService"/> class. /// 创建Kafka主题
/// </summary> /// </summary>
/// <param name="configuration">The configuration.</param> /// <param name="topic"></param>
/// <param name="logger">The logger.</param> /// <param name="numPartitions"></param>
public AdminClientService(IConfiguration configuration, ILogger<AdminClientService> logger) /// <param name="replicationFactor"></param>
/// <returns></returns>
public async Task CreateTopicAsync(string topic, int numPartitions, short replicationFactor)
{
try
{ {
_logger = logger; if (await CheckTopicAsync(topic)) return;
GetInstance(configuration);
}
/// <summary>
/// Gets or sets the instance.
/// </summary>
/// <value>
/// The instance.
/// </value>
public IAdminClient Instance { get; set; } = default;
/// <summary> await Instance.CreateTopicsAsync(new[]
/// Gets the instance.
/// </summary>
/// <param name="configuration">The configuration.</param>
/// <returns></returns>
public IAdminClient GetInstance(IConfiguration configuration)
{
ArgumentNullException.ThrowIfNullOrWhiteSpace(configuration["Kafka:EnableAuthorization"]);
var enableAuthorization = bool.Parse(configuration["Kafka:EnableAuthorization"]!);
var adminClientConfig = new AdminClientConfig()
{ {
BootstrapServers = configuration["Kafka:BootstrapServers"], new TopicSpecification
};
if (enableAuthorization)
{
adminClientConfig.SecurityProtocol = SecurityProtocol.SaslPlaintext;
adminClientConfig.SaslMechanism = SaslMechanism.Plain;
adminClientConfig.SaslUsername = configuration["Kafka:SaslUserName"];
adminClientConfig.SaslPassword = configuration["Kafka:SaslPassword"];
}
Instance = new AdminClientBuilder(adminClientConfig).Build();
return Instance;
}
/// <summary>
/// Checks the topic asynchronous.
/// </summary>
/// <param name="topic">The topic.</param>
/// <returns></returns>
public async Task<bool> CheckTopicAsync(string topic)
{
var metadata = Instance.GetMetadata(TimeSpan.FromSeconds(5));
return await Task.FromResult(metadata.Topics.Exists(a => a.Topic == topic));
}
/// <summary>
/// 判断Kafka主题是否存在
/// </summary>
/// <param name="topic">主题名称</param>
/// <param name="numPartitions">副本数量，不能高于Brokers数量</param>
/// <returns></returns>
public async Task<bool> CheckTopicAsync(string topic,int numPartitions)
{
var metadata = Instance.GetMetadata(TimeSpan.FromSeconds(5));
if(numPartitions > metadata.Brokers.Count)
{
throw new Exception($"{nameof(CheckTopicAsync)} 主题检查时，副本数量大于了节点数量。") ;
}
return await Task.FromResult(metadata.Topics.Exists(a => a.Topic == topic));
}
//// <summary>
/// 创建Kafka主题
/// </summary>
/// <param name="topic">主题名称</param>
/// <param name="numPartitions">主题分区数量</param>
/// <param name="replicationFactor">副本数量，不能高于Brokers数量</param>
/// <returns></returns>
public async Task CreateTopicAsync(string topic, int numPartitions, short replicationFactor)
{
try
{
if (await CheckTopicAsync(topic)) return;
await Instance.CreateTopicsAsync(new[]
{ {
new TopicSpecification Name = topic,
{ NumPartitions = numPartitions,
Name = topic, ReplicationFactor = replicationFactor
NumPartitions = numPartitions,
ReplicationFactor = replicationFactor
}
});
}
catch (CreateTopicsException e)
{
if (e.Results[0].Error.Code != ErrorCode.TopicAlreadyExists)
{
throw;
} }
} });
} }
catch (CreateTopicsException e)
/// <summary>
/// 删除Kafka主题
/// </summary>
/// <param name="topic"></param>
/// <returns></returns>
public async Task DeleteTopicAsync(string topic)
{ {
await Instance.DeleteTopicsAsync(new[] { topic }); if (e.Results[0].Error.Code != ErrorCode.TopicAlreadyExists) throw;
}
/// <summary>
/// 获取Kafka主题列表
/// </summary>
/// <returns></returns>
public async Task<List<string>> ListTopicsAsync()
{
var metadata = Instance.GetMetadata(TimeSpan.FromSeconds(10));
return new List<string>(metadata.Topics.Select(t => t.Topic));
}
/// <summary>
/// 判断Kafka主题是否存在
/// </summary>
/// <param name="topic"></param>
/// <returns></returns>
public async Task<bool> TopicExistsAsync(string topic)
{
var metadata = Instance.GetMetadata(TimeSpan.FromSeconds(10));
return metadata.Topics.Any(t => t.Topic == topic);
}
/// <summary>
/// 检测分区是否存在
/// </summary>
/// <param name="topic"></param>
/// <param name="partitions"></param>
/// <returns></returns>
public Dictionary<int, bool> CheckPartitionsExists(string topic, int[] partitions)
{
var result = new Dictionary<int, bool>();
var metadata = Instance.GetMetadata(topic, TimeSpan.FromSeconds(10));
if (metadata.Topics.Count == 0)
return partitions.ToDictionary(p => p, p => false);
var existingPartitions = metadata.Topics[0].Partitions.Select(p => p.PartitionId).ToHashSet();
return partitions.ToDictionary(p => p, p => existingPartitions.Contains(p));
}
/// <summary>
/// 检测分区是否存在
/// </summary>
/// <param name="topic"></param>
/// <param name="targetPartition"></param>
/// <returns></returns>
public bool CheckPartitionsExist(string topic, int targetPartition)
{
var metadata = Instance.GetMetadata(topic, TimeSpan.FromSeconds(10));
if (metadata.Topics.Count == 0)
return false;
var partitions = metadata.Topics[0].Partitions;
return partitions.Any(p => p.PartitionId == targetPartition);
}
/// <summary>
/// 获取主题的分区数量
/// </summary>
/// <param name="topic"></param>
/// <returns></returns>
public int GetTopicPartitionsNum(string topic)
{
var metadata = Instance.GetMetadata(topic, TimeSpan.FromSeconds(10));
if (metadata.Topics.Count == 0)
return 0;
return metadata.Topics[0].Partitions.Count;
}
public void Dispose()
{
Instance?.Dispose();
} }
} }
/// <summary>
/// 删除Kafka主题
/// </summary>
/// <param name="topic"></param>
/// <returns></returns>
public async Task DeleteTopicAsync(string topic)
{
await Instance.DeleteTopicsAsync(new[] { topic });
}
/// <summary>
/// 获取Kafka主题列表
/// </summary>
/// <returns></returns>
public async Task<List<string>> ListTopicsAsync()
{
var metadata = Instance.GetMetadata(TimeSpan.FromSeconds(10));
return new List<string>(metadata.Topics.Select(t => t.Topic));
}
/// <summary>
/// 判断Kafka主题是否存在
/// </summary>
/// <param name="topic"></param>
/// <returns></returns>
public async Task<bool> TopicExistsAsync(string topic)
{
var metadata = Instance.GetMetadata(TimeSpan.FromSeconds(10));
return metadata.Topics.Any(t => t.Topic == topic);
}
/// <summary>
/// 检测分区是否存在
/// </summary>
/// <param name="topic"></param>
/// <param name="partitions"></param>
/// <returns></returns>
public Dictionary<int, bool> CheckPartitionsExists(string topic, int[] partitions)
{
var result = new Dictionary<int, bool>();
var metadata = Instance.GetMetadata(topic, TimeSpan.FromSeconds(10));
if (metadata.Topics.Count == 0)
return partitions.ToDictionary(p => p, p => false);
var existingPartitions = metadata.Topics[0].Partitions.Select(p => p.PartitionId).ToHashSet();
return partitions.ToDictionary(p => p, p => existingPartitions.Contains(p));
}
/// <summary>
/// 检测分区是否存在
/// </summary>
/// <param name="topic"></param>
/// <param name="targetPartition"></param>
/// <returns></returns>
public bool CheckPartitionsExist(string topic, int targetPartition)
{
var metadata = Instance.GetMetadata(topic, TimeSpan.FromSeconds(10));
if (metadata.Topics.Count == 0)
return false;
var partitions = metadata.Topics[0].Partitions;
return partitions.Any(p => p.PartitionId == targetPartition);
}
/// <summary>
/// 获取主题的分区数量
/// </summary>
/// <param name="topic"></param>
/// <returns></returns>
public int GetTopicPartitionsNum(string topic)
{
var metadata = Instance.GetMetadata(topic, TimeSpan.FromSeconds(10));
if (metadata.Topics.Count == 0)
return 0;
return metadata.Topics[0].Partitions.Count;
}
public void Dispose()
{
Instance?.Dispose();
}
/// <summary>
/// Gets the instance.
/// </summary>
/// <param name="configuration">The configuration.</param>
/// <returns></returns>
public IAdminClient GetInstance(IConfiguration configuration)
{
ArgumentException.ThrowIfNullOrWhiteSpace(configuration["Kafka:EnableAuthorization"]);
var enableAuthorization = bool.Parse(configuration["Kafka:EnableAuthorization"]!);
var adminClientConfig = new AdminClientConfig
{
BootstrapServers = configuration["Kafka:BootstrapServers"]
};
if (enableAuthorization)
{
adminClientConfig.SecurityProtocol = SecurityProtocol.SaslPlaintext;
adminClientConfig.SaslMechanism = SaslMechanism.Plain;
adminClientConfig.SaslUsername = configuration["Kafka:SaslUserName"];
adminClientConfig.SaslPassword = configuration["Kafka:SaslPassword"];
}
return new AdminClientBuilder(adminClientConfig).Build();
}
/// <summary>
/// Checks the topic asynchronous.
/// </summary>
/// <param name="topic">The topic.</param>
/// <returns></returns>
public async Task<bool> CheckTopicAsync(string topic)
{
var metadata = Instance.GetMetadata(TimeSpan.FromSeconds(5));
return await Task.FromResult(metadata.Topics.Exists(a => a.Topic == topic));
}
/// <summary>
/// 判断Kafka主题是否存在
/// </summary>
/// <param name="topic">主题名称</param>
/// <param name="numPartitions">副本数量，不能高于Brokers数量</param>
/// <returns></returns>
public async Task<bool> CheckTopicAsync(string topic, int numPartitions)
{
var metadata = Instance.GetMetadata(TimeSpan.FromSeconds(5));
if (numPartitions > metadata.Brokers.Count)
throw new Exception($"{nameof(CheckTopicAsync)} 主题检查时，副本数量大于了节点数量。");
return await Task.FromResult(metadata.Topics.Exists(a => a.Topic == topic));
}
} }

118
modules/JiShe.CollectBus.Kafka/Attributes/KafkaSubscribeAttribute.cs
View File

@ -1,68 +1,60 @@
using System; namespace JiShe.CollectBus.Kafka.Attributes;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace JiShe.CollectBus.Kafka.Attributes [AttributeUsage(AttributeTargets.Method)]
public class KafkaSubscribeAttribute : Attribute
{ {
[AttributeUsage(AttributeTargets.Method)] /// <summary>
public class KafkaSubscribeAttribute : Attribute /// 订阅主题
/// </summary>
/// <param name="batchTimeout"></param>
public KafkaSubscribeAttribute(string topic)
{ {
/// <summary> Topic = topic;
/// 订阅的主题
/// </summary>
public string Topic { get; set; } = null!;
/// <summary>
/// 分区
/// </summary>
public int Partition { get; set; } = -1;
/// <summary>
/// 消费者组
/// </summary>
public string? GroupId { get; set; } = null;//"default"
/// <summary>
/// 任务数(默认是多少个分区多少个任务)
/// 如设置订阅指定Partition则任务数始终为1
/// </summary>
public int TaskCount { get; set; } = -1;
/// <summary>
/// 批量处理数量
/// </summary>
public int BatchSize { get; set; } = 100;
/// <summary>
/// 是否启用批量处理
/// </summary>
public bool EnableBatch { get; set; } = false;
/// <summary>
/// 批次超时时间
/// 格式:("00:05:00")
/// </summary>
public TimeSpan? BatchTimeout { get; set; }=null;
/// <summary>
/// 订阅主题
/// </summary>
/// <param name="batchTimeout"></param>
public KafkaSubscribeAttribute(string topic)
{
this.Topic = topic;
}
/// <summary>
/// 订阅主题
/// </summary>
public KafkaSubscribeAttribute(string topic, int partition)
{
this.Topic = topic;
this.Partition = partition;
}
} }
/// <summary>
/// 订阅主题
/// </summary>
public KafkaSubscribeAttribute(string topic, int partition)
{
Topic = topic;
Partition = partition;
}
/// <summary>
/// 订阅的主题
/// </summary>
public string Topic { get; set; } = null!;
/// <summary>
/// 分区
/// </summary>
public int Partition { get; set; } = -1;
/// <summary>
/// 消费者组
/// </summary>
public string? GroupId { get; set; } = null; //"default"
/// <summary>
/// 任务数(默认是多少个分区多少个任务)
/// 如设置订阅指定Partition则任务数始终为1
/// </summary>
public int TaskCount { get; set; } = -1;
/// <summary>
/// 批量处理数量
/// </summary>
public int BatchSize { get; set; } = 100;
/// <summary>
/// 是否启用批量处理
/// </summary>
public bool EnableBatch { get; set; } = false;
/// <summary>
/// 批次超时时间
/// 格式:("00:05:00")
/// </summary>
public TimeSpan? BatchTimeout { get; set; } = null;
} }

41
modules/JiShe.CollectBus.Kafka/Attributes/TopicAttribute.cs
View File

@ -1,29 +1,22 @@
using System; namespace JiShe.CollectBus.Kafka.Attributes;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace JiShe.CollectBus.Kafka.Attributes [AttributeUsage(AttributeTargets.Class, Inherited = false)]
public class TopicAttribute : Attribute
{ {
[AttributeUsage(AttributeTargets.Class, Inherited = false)] /// <summary>
public class TopicAttribute: Attribute /// Initializes a new instance of the <see cref="TopicAttribute" /> class.
/// </summary>
/// <param name="name">The name.</param>
public TopicAttribute(string name = "Default")
{ {
/// <summary> Name = name;
/// Initializes a new instance of the <see cref="TopicAttribute"/> class.
/// </summary>
/// <param name="name">The name.</param>
public TopicAttribute(string name = "Default")
{
Name = name;
}
/// <summary>
/// Gets or sets the name.
/// </summary>
/// <value>
/// The name.
/// </value>
public string Name { get; set; }
} }
/// <summary>
/// Gets or sets the name.
/// </summary>
/// <value>
/// The name.
/// </value>
public string Name { get; set; }
} }

13
modules/JiShe.CollectBus.Kafka/Consumer/ConsumerService.cs
View File

@ -13,15 +13,18 @@ namespace JiShe.CollectBus.Kafka.Consumer
public class ConsumerService : IConsumerService, IDisposable public class ConsumerService : IConsumerService, IDisposable
{ {
private readonly ILogger<ConsumerService> _logger; private readonly ILogger<ConsumerService> _logger;
private readonly IConfiguration _configuration;
private readonly ConcurrentDictionary<Type, (object Consumer, CancellationTokenSource CTS)> private readonly ConcurrentDictionary<Type, (object Consumer, CancellationTokenSource CTS)>
_consumerStore = new(); _consumerStore = new();
private readonly KafkaOptionConfig _kafkaOptionConfig; private readonly KafkaOptionConfig _kafkaOptionConfig;
private class KafkaConsumer<TKey, TValue> where TKey : notnull where TValue : class { } private class KafkaConsumer<TKey, TValue> where TKey : notnull where TValue : class { }
public ConsumerService(IConfiguration configuration, ILogger<ConsumerService> logger, IOptions<KafkaOptionConfig> kafkaOptionConfig) /// <summary>
/// ConsumerService
/// </summary>
/// <param name="logger"></param>
/// <param name="kafkaOptionConfig"></param>
public ConsumerService(ILogger<ConsumerService> logger, IOptions<KafkaOptionConfig> kafkaOptionConfig)
{ {
_configuration = configuration;
_logger = logger; _logger = logger;
_kafkaOptionConfig = kafkaOptionConfig.Value; _kafkaOptionConfig = kafkaOptionConfig.Value;
} }
@ -165,10 +168,10 @@ namespace JiShe.CollectBus.Kafka.Consumer
/// <summary> /// <summary>
/// 订阅消息 /// 订阅消息
/// </summary> /// </summary>
/// <typeparam name="TKey"></typeparam>
/// <typeparam name="TValue"></typeparam> /// <typeparam name="TValue"></typeparam>
/// <param name="topics"></param> /// <param name="topics"></param>
/// <param name="messageHandler"></param> /// <param name="messageHandler"></param>
/// <param name="groupId"></param>
/// <returns></returns> /// <returns></returns>
public async Task SubscribeAsync<TValue>(string[] topics, Func<TValue, Task<bool>> messageHandler, string? groupId) where TValue : class public async Task SubscribeAsync<TValue>(string[] topics, Func<TValue, Task<bool>> messageHandler, string? groupId) where TValue : class
{ {
@ -387,7 +390,7 @@ namespace JiShe.CollectBus.Kafka.Consumer
/// <param name="consumeTimeout">消费等待时间</param> /// <param name="consumeTimeout">消费等待时间</param>
public async Task SubscribeBatchAsync<TValue>(string topic, Func<List<TValue>, Task<bool>> messageBatchHandler, string? groupId = null, int batchSize = 100, TimeSpan? batchTimeout = null, TimeSpan? consumeTimeout = null) where TValue : class public async Task SubscribeBatchAsync<TValue>(string topic, Func<List<TValue>, Task<bool>> messageBatchHandler, string? groupId = null, int batchSize = 100, TimeSpan? batchTimeout = null, TimeSpan? consumeTimeout = null) where TValue : class
{ {
await SubscribeBatchAsync<TValue>(new[] { topic }, messageBatchHandler, groupId, batchSize, batchTimeout, consumeTimeout); await SubscribeBatchAsync(new[] { topic }, messageBatchHandler, groupId, batchSize, batchTimeout, consumeTimeout);
} }

72
modules/JiShe.CollectBus.Kafka/Consumer/IConsumerService.cs
View File

@ -1,46 +1,50 @@
using Confluent.Kafka; namespace JiShe.CollectBus.Kafka.Consumer;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace JiShe.CollectBus.Kafka.Consumer public interface IConsumerService
{ {
public interface IConsumerService Task SubscribeAsync<TKey, TValue>(string topic, Func<TKey, TValue, Task<bool>> messageHandler,
{ string? groupId = null) where TKey : notnull where TValue : class;
Task SubscribeAsync<TKey, TValue>(string topic, Func<TKey, TValue, Task<bool>> messageHandler, string? groupId=null) where TKey : notnull where TValue : class;
/// <summary> /// <summary>
/// 订阅消息 /// 订阅消息
/// </summary> /// </summary>
/// <typeparam name="TValue"></typeparam> /// <typeparam name="TValue"></typeparam>
/// <param name="topic"></param> /// <param name="topic"></param>
/// <param name="messageHandler"></param> /// <param name="messageHandler"></param>
/// <returns></returns> /// <returns></returns>
Task SubscribeAsync<TValue>(string topic, Func<TValue, Task<bool>> messageHandler, string? groupId = null) where TValue : class; Task SubscribeAsync<TValue>(string topic, Func<TValue, Task<bool>> messageHandler, string? groupId = null)
where TValue : class;
Task SubscribeAsync<TKey, TValue>(string[] topics, Func<TKey, TValue, Task<bool>> messageHandler, string? groupId) where TKey : notnull where TValue : class; Task SubscribeAsync<TKey, TValue>(string[] topics, Func<TKey, TValue, Task<bool>> messageHandler, string? groupId)
where TKey : notnull where TValue : class;
/// <summary> /// <summary>
/// 订阅消息 /// 订阅消息
/// </summary> /// </summary>
/// <typeparam name="TKey"></typeparam> /// <typeparam name="TKey"></typeparam>
/// <typeparam name="TValue"></typeparam> /// <typeparam name="TValue"></typeparam>
/// <param name="topics"></param> /// <param name="topics"></param>
/// <param name="messageHandler"></param> /// <param name="messageHandler"></param>
/// <returns></returns> /// <returns></returns>
Task SubscribeAsync<TValue>(string[] topics, Func<TValue, Task<bool>> messageHandler, string? groupId = null) where TValue : class; Task SubscribeAsync<TValue>(string[] topics, Func<TValue, Task<bool>> messageHandler, string? groupId = null)
where TValue : class;
Task SubscribeBatchAsync<TKey, TValue>(string[] topics, Func<List<TValue>, Task<bool>> messageBatchHandler, string? groupId = null, int batchSize = 100, TimeSpan? batchTimeout = null) where TKey : notnull where TValue : class; Task SubscribeBatchAsync<TKey, TValue>(string[] topics, Func<List<TValue>, Task<bool>> messageBatchHandler,
string? groupId = null, int batchSize = 100, TimeSpan? batchTimeout = null)
where TKey : notnull where TValue : class;
Task SubscribeBatchAsync<TKey, TValue>(string topic, Func<List<TValue>, Task<bool>> messageBatchHandler, string? groupId = null, int batchSize = 100, TimeSpan? batchTimeout = null) where TKey : notnull where TValue : class; Task SubscribeBatchAsync<TKey, TValue>(string topic, Func<List<TValue>, Task<bool>> messageBatchHandler,
string? groupId = null, int batchSize = 100, TimeSpan? batchTimeout = null)
where TKey : notnull where TValue : class;
Task SubscribeBatchAsync<TValue>(string topic, Func<List<TValue>, Task<bool>> messageBatchHandler, string? groupId = null, int batchSize = 100, TimeSpan? batchTimeout = null, TimeSpan? consumeTimeout = null) where TValue : class; Task SubscribeBatchAsync<TValue>(string topic, Func<List<TValue>, Task<bool>> messageBatchHandler,
string? groupId = null, int batchSize = 100, TimeSpan? batchTimeout = null, TimeSpan? consumeTimeout = null)
where TValue : class;
Task SubscribeBatchAsync<TValue>(string[] topics, Func<List<TValue>, Task<bool>> messageBatchHandler, string? groupId = null, int batchSize = 100, TimeSpan? batchTimeout = null, TimeSpan? consumeTimeout = null) where TValue : class; Task SubscribeBatchAsync<TValue>(string[] topics, Func<List<TValue>, Task<bool>> messageBatchHandler,
string? groupId = null, int batchSize = 100, TimeSpan? batchTimeout = null, TimeSpan? consumeTimeout = null)
where TValue : class;
void Unsubscribe<TKey, TValue>() where TKey : notnull where TValue : class; void Unsubscribe<TKey, TValue>() where TKey : notnull where TValue : class;
}
} }

36
modules/JiShe.CollectBus.Kafka/Internal/HeadersFilter.cs
View File

@ -1,30 +1,22 @@
using Confluent.Kafka; using Confluent.Kafka;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace JiShe.CollectBus.Kafka.Internal namespace JiShe.CollectBus.Kafka.Internal;
/// <summary>
/// 消息头过滤器
/// </summary>
public class HeadersFilter : Dictionary<string, byte[]>
{ {
/// <summary> /// <summary>
/// 消息头过滤器 /// 判断Headers是否匹配
/// </summary> /// </summary>
public class HeadersFilter : Dictionary<string, byte[]> /// <param name="headers"></param>
/// <returns></returns>
public bool Match(Headers headers)
{ {
/// <summary> foreach (var kvp in this)
/// 判断Headers是否匹配 if (!headers.TryGetLastBytes(kvp.Key, out var value) || !value.SequenceEqual(kvp.Value))
/// </summary> return false;
/// <param name="headers"></param> return true;
/// <returns></returns>
public bool Match(Headers headers)
{
foreach (var kvp in this)
{
if (!headers.TryGetLastBytes(kvp.Key, out var value) || !value.SequenceEqual(kvp.Value))
return false;
}
return true;
}
} }
} }

23
modules/JiShe.CollectBus.Kafka/Internal/IKafkaSubscribe.cs
View File

@ -1,18 +1,11 @@
using System; namespace JiShe.CollectBus.Kafka.Internal;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace JiShe.CollectBus.Kafka.Internal /// <summary>
/// Kafka订阅者
/// <para>
/// 订阅者需要继承此接口并需要依赖注入，并使用<see cref="KafkaSubscribeAttribute" />标记
/// </para>
/// </summary>
public interface IKafkaSubscribe
{ {
/// <summary>
/// Kafka订阅者
/// <para>
/// 订阅者需要继承此接口并需要依赖注入，并使用<see cref="KafkaSubscribeAttribute"/>标记
/// </para>
/// </summary>
public interface IKafkaSubscribe
{
}
} }

27
modules/JiShe.CollectBus.Kafka/Internal/ISubscribeAck.cs
View File

@ -1,21 +1,14 @@
using System; namespace JiShe.CollectBus.Kafka.Internal;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace JiShe.CollectBus.Kafka.Internal public interface ISubscribeAck
{ {
public interface ISubscribeAck /// <summary>
{ /// 是否成功标记
/// <summary> /// </summary>
/// 是否成功标记 bool Ack { get; set; }
/// </summary>
bool Ack { get; set; }
/// <summary> /// <summary>
/// 消息 /// 消息
/// </summary> /// </summary>
string? Msg { get; set; } string? Msg { get; set; }
}
} }

101
modules/JiShe.CollectBus.Kafka/Internal/KafkaOptionConfig.cs
View File

@ -1,68 +1,61 @@
using Confluent.Kafka; using Confluent.Kafka;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace JiShe.CollectBus.Kafka.Internal namespace JiShe.CollectBus.Kafka.Internal;
public class KafkaOptionConfig
{ {
public class KafkaOptionConfig /// <summary>
{ /// kafka地址
/// <summary> /// </summary>
/// kafka地址 public string BootstrapServers { get; set; } = null!;
/// </summary>
public string BootstrapServers { get; set; } = null!;
/// <summary> /// <summary>
/// 服务器标识 /// 服务器标识
/// </summary> /// </summary>
public string ServerTagName { get; set; }= "KafkaFilterKey"; public string ServerTagName { get; set; } = "KafkaFilterKey";
/// <summary> /// <summary>
/// kafka主题副本数量 /// kafka主题副本数量
/// </summary> /// </summary>
public short KafkaReplicationFactor { get; set; } public short KafkaReplicationFactor { get; set; }
/// <summary> /// <summary>
/// kafka主题分区数量 /// kafka主题分区数量
/// </summary> /// </summary>
public int NumPartitions { get; set; } public int NumPartitions { get; set; }
/// <summary> /// <summary>
/// 是否开启过滤器 /// 是否开启过滤器
/// </summary> /// </summary>
public bool EnableFilter { get; set; }= true; public bool EnableFilter { get; set; } = true;
/// <summary> /// <summary>
/// 是否开启认证 /// 是否开启认证
/// </summary> /// </summary>
public bool EnableAuthorization { get; set; } = false; public bool EnableAuthorization { get; set; } = false;
/// <summary> /// <summary>
/// 安全协议 /// 安全协议
/// </summary> /// </summary>
public SecurityProtocol SecurityProtocol { get; set; } = SecurityProtocol.SaslPlaintext; public SecurityProtocol SecurityProtocol { get; set; } = SecurityProtocol.SaslPlaintext;
/// <summary> /// <summary>
/// 认证方式 /// 认证方式
/// </summary> /// </summary>
public SaslMechanism SaslMechanism { get; set; }= SaslMechanism.Plain; public SaslMechanism SaslMechanism { get; set; } = SaslMechanism.Plain;
/// <summary> /// <summary>
/// 用户名 /// 用户名
/// </summary> /// </summary>
public string? SaslUserName { get; set; } public string? SaslUserName { get; set; }
/// <summary> /// <summary>
/// 密码 /// 密码
/// </summary> /// </summary>
public string? SaslPassword { get; set; } public string? SaslPassword { get; set; }
/// <summary> /// <summary>
/// 首次采集时间 /// 首次采集时间
/// </summary> /// </summary>
public DateTime FirstCollectionTime { get; set; } public DateTime FirstCollectionTime { get; set; }
}
} }

178
modules/JiShe.CollectBus.Kafka/Internal/ReflectionHelper.cs
View File

@ -1,113 +1,103 @@
using Newtonsoft.Json; using System.Collections;
using System;
using System.Collections;
using System.Collections.Generic;
using System.Linq;
using System.Reflection; using System.Reflection;
using System.Text;
using System.Threading.Tasks;
namespace JiShe.CollectBus.Kafka.Internal namespace JiShe.CollectBus.Kafka.Internal;
/// <summary>
/// 反射辅助类
/// </summary>
public static class ReflectionHelper
{ {
/// <summary> /// <summary>
/// 反射辅助类 /// 集合类型
/// Item1：参数类型
/// Item2：集合元素类型
/// </summary> /// </summary>
public static class ReflectionHelper public static Tuple<Type, Type?> GetParameterTypeInfo(this MethodInfo method, int parameterIndex = 0)
{ {
/// <summary> // 参数校验
///集合类型 if (method == null) throw new ArgumentNullException(nameof(method));
///Item1：参数类型 var parameters = method.GetParameters();
///Item2：集合元素类型 if (parameterIndex < 0 || parameterIndex >= parameters.Length)
/// </summary> throw new ArgumentOutOfRangeException(nameof(parameterIndex));
public static Tuple<Type,Type?> GetParameterTypeInfo(this MethodInfo method, int parameterIndex=0)
{
// 参数校验
if (method == null) throw new ArgumentNullException(nameof(method));
var parameters = method.GetParameters();
if (parameterIndex < 0 || parameterIndex >= parameters.Length)
throw new ArgumentOutOfRangeException(nameof(parameterIndex));
ParameterInfo param = parameters[parameterIndex]; var param = parameters[parameterIndex];
Type paramType = param.ParameterType; var paramType = param.ParameterType;
Type? elementType = null; Type? elementType = null;
// 判断是否是集合类型（排除字符串） // 判断是否是集合类型（排除字符串）
if (paramType != typeof(string) && IsEnumerableType(paramType)) if (paramType != typeof(string) && IsEnumerableType(paramType))
{ elementType = GetEnumerableElementType(paramType);
elementType = GetEnumerableElementType(paramType);
}
return Tuple.Create(paramType, elementType); return Tuple.Create(paramType, elementType);
}
} /// <summary>
/// 判断是否是集合类型（排除字符串）
/// </summary>
public static bool IsEnumerableType(this Type type)
{
return type.IsArray
|| (type.IsGenericType && type.GetInterfaces()
.Any(t => t.IsGenericType
&& t.GetGenericTypeDefinition() == typeof(IEnumerable<>)))
|| type.GetInterfaces().Any(t => t == typeof(IEnumerable));
}
/// <summary> /// <summary>
/// 判断是否是集合类型（排除字符串） /// 获取集合元素的类型
/// </summary> /// </summary>
public static bool IsEnumerableType(this Type type) public static Type? GetEnumerableElementType(this Type type)
{ {
return type.IsArray // 处理数组类型
|| (type.IsGenericType && type.GetInterfaces() if (type.IsArray)
.Any(t => t.IsGenericType return type.GetElementType();
&& t.GetGenericTypeDefinition() == typeof(IEnumerable<>)))
|| type.GetInterfaces().Any(t => t == typeof(System.Collections.IEnumerable));
}
/// <summary> // 处理直接实现IEnumerable<T>的类型（如IEnumerable<int>本身）
/// 获取集合元素的类型 if (type.IsGenericType && type.GetGenericTypeDefinition() == typeof(IEnumerable<>))
/// </summary> return type.GetGenericArguments()[0];
public static Type? GetEnumerableElementType(this Type type)
{
// 处理数组类型
if (type.IsArray)
return type.GetElementType();
// 处理直接实现IEnumerable<T>的类型（如IEnumerable<int>本身） // 处理通过接口实现IEnumerable<T>的泛型集合（如List<T>）
if (type.IsGenericType && type.GetGenericTypeDefinition() == typeof(IEnumerable<>)) var genericEnumerable = type.GetInterfaces()
return type.GetGenericArguments()[0]; .FirstOrDefault(t => t.IsGenericType
&& t.GetGenericTypeDefinition() == typeof(IEnumerable<>));
if (genericEnumerable != null)
return genericEnumerable.GetGenericArguments()[0];
// 处理通过接口实现IEnumerable<T>的泛型集合（如List<T>） // 处理非泛型集合类型（如 ArrayList）
var genericEnumerable = type.GetInterfaces() if (typeof(IEnumerable).IsAssignableFrom(type) && type == typeof(ArrayList))
.FirstOrDefault(t => t.IsGenericType return typeof(ArrayList);
&& t.GetGenericTypeDefinition() == typeof(IEnumerable<>)); // 返回null表示无法确定元素类型
if (genericEnumerable != null) return null;
return genericEnumerable.GetGenericArguments()[0]; }
// 处理非泛型集合类型（如 ArrayList）
if (typeof(IEnumerable).IsAssignableFrom(type) && type == typeof(ArrayList))
return typeof(ArrayList);
// 返回null表示无法确定元素类型
return null;
}
// <summary> /// <summary>
/// 判断是否使用强转换 /// 判断是否使用强转换
/// </summary> /// </summary>
/// <param name="targetType">目标类型</param> /// <param name="targetType"></param>
/// <returns></returns> /// <returns></returns>
public static bool IsConvertType(this Type targetType) public static bool IsConvertType(this Type targetType)
{ {
// 处理可空类型 // 处理可空类型
Type underlyingType = Nullable.GetUnderlyingType(targetType) ?? targetType; var underlyingType = Nullable.GetUnderlyingType(targetType) ?? targetType;
// 情况1：值类型或基元类型（如 int、DateTime） // 情况1：值类型或基元类型（如 int、DateTime）
if (underlyingType.IsValueType || underlyingType.IsPrimitive) if (underlyingType.IsValueType || underlyingType.IsPrimitive)
return true; return true;
// 情况2：字符串类型直接赋值 // 情况2：字符串类型直接赋值
else if (underlyingType == typeof(string)) if (underlyingType == typeof(string))
return true; return true;
// 情况3：枚举类型处理 // 情况3：枚举类型处理
//else if (underlyingType.IsEnum) //else if (underlyingType.IsEnum)
//{ //{
// if (Enum.IsDefined(underlyingType, msg)) // if (Enum.IsDefined(underlyingType, msg))
// { // {
// convertedValue = Enum.Parse(underlyingType, msg.ToString()); // convertedValue = Enum.Parse(underlyingType, msg.ToString());
// return true; // return true;
// } // }
// return false; // return false;
//} //}
return false; return false;
}
} }
} }

121
modules/JiShe.CollectBus.Kafka/Internal/SubscribeResult.cs
View File

@ -1,75 +1,62 @@
using Confluent.Kafka; namespace JiShe.CollectBus.Kafka.Internal;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using static System.Runtime.InteropServices.JavaScript.JSType;
namespace JiShe.CollectBus.Kafka.Internal public class SubscribeResult : ISubscribeAck
{ {
public class SubscribeResult: ISubscribeAck /// <summary>
/// 是否成功
/// </summary>
public bool Ack { get; set; }
/// <summary>
/// 消息
/// </summary>
public string? Msg { get; set; }
/// <summary>
/// 成功
/// </summary>
/// <param name="msg">消息</param>
public SubscribeResult Success(string? msg = null)
{ {
/// <summary> Ack = true;
/// 是否成功 Msg = msg;
/// </summary> return this;
public bool Ack { get; set; }
/// <summary>
/// 消息
/// </summary>
public string? Msg { get; set; }
/// <summary>
/// 成功
/// </summary>
/// <param name="msg">消息</param>
public SubscribeResult Success(string? msg = null)
{
Ack = true;
Msg = msg;
return this;
}
/// <summary>
/// 失败
/// </summary>
/// <param name="code"></param>
/// <param name="msg"></param>
/// <param name="data"></param>
/// <returns></returns>
public SubscribeResult Fail(string? msg = null)
{
Msg = msg;
Ack = false;
return this;
}
} }
public static partial class SubscribeAck /// <summary>
/// 失败
/// </summary>
/// <param name="msg"></param>
/// <returns></returns>
public SubscribeResult Fail(string? msg = null)
{ {
Msg = msg;
/// <summary> Ack = false;
/// 成功 return this;
/// </summary> }
/// <param name="msg">消息</param> }
/// <returns></returns>
public static ISubscribeAck Success(string? msg = null) public static class SubscribeAck
{ {
return new SubscribeResult().Success(msg); /// <summary>
} /// 成功
/// </summary>
/// <param name="msg">消息</param>
/// <summary> /// <returns></returns>
/// 失败 public static ISubscribeAck Success(string? msg = null)
/// </summary> {
/// <param name="msg">消息</param> return new SubscribeResult().Success(msg);
/// <returns></returns> }
public static ISubscribeAck Fail(string? msg = null)
{
return new SubscribeResult().Fail(msg); /// <summary>
} /// 失败
/// </summary>
/// <param name="msg">消息</param>
/// <returns></returns>
public static ISubscribeAck Fail(string? msg = null)
{
return new SubscribeResult().Fail(msg);
} }
} }

36
modules/JiShe.CollectBus.Kafka/KafkaSubcribesExtensions.cs → modules/JiShe.CollectBus.Kafka/KafkaSubscribeExtensions.cs
View File

@ -8,26 +8,17 @@ using JiShe.CollectBus.Kafka.Consumer;
using JiShe.CollectBus.Kafka.Internal; using JiShe.CollectBus.Kafka.Internal;
using JiShe.CollectBus.Kafka.Serialization; using JiShe.CollectBus.Kafka.Serialization;
using Microsoft.AspNetCore.Builder; using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Mvc.Abstractions;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options; using Microsoft.Extensions.Options;
using System;
using System.Collections; using System.Collections;
using System.Collections.Generic;
using System.ComponentModel; using System.ComponentModel;
using System.Linq.Expressions;
using System.Reflection; using System.Reflection;
using System.Text.Json;
using System.Threading.Tasks;
using YamlDotNet.Core.Tokens;
using static System.Runtime.InteropServices.JavaScript.JSType;
namespace JiShe.CollectBus.Kafka namespace JiShe.CollectBus.Kafka
{ {
public static class KafkaSubcribesExtensions public static class KafkaSubscribeExtensions
{ {
public static void UseInitKafkaTopic(this IServiceProvider provider) public static void UseInitKafkaTopic(this IServiceProvider provider)
@ -36,7 +27,7 @@ namespace JiShe.CollectBus.Kafka
var kafkaAdminClient = provider.GetRequiredService<IAdminClientService>(); var kafkaAdminClient = provider.GetRequiredService<IAdminClientService>();
var kafkaOptions = provider.GetRequiredService<IOptions<KafkaOptionConfig>>(); var kafkaOptions = provider.GetRequiredService<IOptions<KafkaOptionConfig>>();
List<string> topics = ProtocolConstExtensions.GetAllTopicNamesByIssued(); var topics = ProtocolConstExtensions.GetAllTopicNamesByIssued();
topics.AddRange(ProtocolConstExtensions.GetAllTopicNamesByReceived()); topics.AddRange(ProtocolConstExtensions.GetAllTopicNamesByReceived());
foreach (var item in topics) foreach (var item in topics)
@ -48,8 +39,6 @@ namespace JiShe.CollectBus.Kafka
/// <summary> /// <summary>
/// 添加Kafka订阅 /// 添加Kafka订阅
/// </summary> /// </summary>
/// <param name="app"></param>
/// <param name="assembly"></param>
public static void UseKafkaSubscribe(this IServiceProvider provider) public static void UseKafkaSubscribe(this IServiceProvider provider)
{ {
var lifetime = provider.GetRequiredService<IHostApplicationLifetime>(); var lifetime = provider.GetRequiredService<IHostApplicationLifetime>();
@ -57,8 +46,8 @@ namespace JiShe.CollectBus.Kafka
lifetime.ApplicationStarted.Register(() => lifetime.ApplicationStarted.Register(() =>
{ {
var logger = provider.GetRequiredService<ILogger<CollectBusKafkaModule>>(); var logger = provider.GetRequiredService<ILogger<CollectBusKafkaModule>>();
int threadCount = 0; var threadCount = 0;
int topicCount = 0; var topicCount = 0;
var assemblyPath = Path.GetDirectoryName(Assembly.GetEntryAssembly()?.Location); var assemblyPath = Path.GetDirectoryName(Assembly.GetEntryAssembly()?.Location);
if (string.IsNullOrWhiteSpace(assemblyPath)) if (string.IsNullOrWhiteSpace(assemblyPath))
{ {
@ -98,6 +87,9 @@ namespace JiShe.CollectBus.Kafka
}); });
} }
/// <summary>
/// 添加Kafka订阅
/// </summary>
public static void UseKafkaSubscribersAsync(this IApplicationBuilder app, Assembly assembly) public static void UseKafkaSubscribersAsync(this IApplicationBuilder app, Assembly assembly)
{ {
var provider = app.ApplicationServices; var provider = app.ApplicationServices;
@ -134,8 +126,6 @@ namespace JiShe.CollectBus.Kafka
/// <summary> /// <summary>
/// 构建Kafka订阅 /// 构建Kafka订阅
/// </summary> /// </summary>
/// <param name="subscribe"></param>
/// <param name="provider"></param>
private static Tuple<int, int> BuildKafkaSubscribe(object subscribe, IServiceProvider provider, ILogger<CollectBusKafkaModule> logger, KafkaOptionConfig kafkaOptionConfig) private static Tuple<int, int> BuildKafkaSubscribe(object subscribe, IServiceProvider provider, ILogger<CollectBusKafkaModule> logger, KafkaOptionConfig kafkaOptionConfig)
{ {
var subscribedMethods = subscribe.GetType().GetMethods() var subscribedMethods = subscribe.GetType().GetMethods()
@ -169,11 +159,6 @@ namespace JiShe.CollectBus.Kafka
/// <summary> /// <summary>
/// 启动后台消费线程 /// 启动后台消费线程
/// </summary> /// </summary>
/// <param name="config"></param>
/// <param name="attr"></param>
/// <param name="method"></param>
/// <param name="consumerInstance"></param>
/// <returns></returns>
private static async Task StartConsumerAsync(IServiceProvider provider, KafkaSubscribeAttribute attr, MethodInfo method, object subscribe, ILogger<CollectBusKafkaModule> logger) private static async Task StartConsumerAsync(IServiceProvider provider, KafkaSubscribeAttribute attr, MethodInfo method, object subscribe, ILogger<CollectBusKafkaModule> logger)
{ {
var consumerService = provider.GetRequiredService<IConsumerService>(); var consumerService = provider.GetRequiredService<IConsumerService>();
@ -225,10 +210,6 @@ namespace JiShe.CollectBus.Kafka
/// <summary> /// <summary>
/// 处理消息 /// 处理消息
/// </summary> /// </summary>
/// <param name="message"></param>
/// <param name="method"></param>
/// <param name="subscribe"></param>
/// <returns></returns>
private static async Task<bool> ProcessMessageAsync(List<dynamic> messages, MethodInfo method, object subscribe) private static async Task<bool> ProcessMessageAsync(List<dynamic> messages, MethodInfo method, object subscribe)
{ {
var parameters = method.GetParameters(); var parameters = method.GetParameters();
@ -351,9 +332,6 @@ namespace JiShe.CollectBus.Kafka
} }
return false; return false;
} }
} }

5
modules/JiShe.CollectBus.Kafka/Producer/IProducerService.cs
View File

@ -1,9 +1,4 @@
using Confluent.Kafka; using Confluent.Kafka;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace JiShe.CollectBus.Kafka.Producer namespace JiShe.CollectBus.Kafka.Producer
{ {

7
modules/JiShe.CollectBus.Kafka/Producer/ProducerService.cs
View File

@ -23,6 +23,13 @@ namespace JiShe.CollectBus.Kafka.Producer
private readonly ConcurrentDictionary<Type, object> _producerCache = new(); private readonly ConcurrentDictionary<Type, object> _producerCache = new();
private class KafkaProducer<TKey, TValue> where TKey : notnull where TValue : class { } private class KafkaProducer<TKey, TValue> where TKey : notnull where TValue : class { }
private readonly KafkaOptionConfig _kafkaOptionConfig; private readonly KafkaOptionConfig _kafkaOptionConfig;
/// <summary>
/// ProducerService
/// </summary>
/// <param name="configuration"></param>
/// <param name="logger"></param>
/// <param name="kafkaOptionConfig"></param>
public ProducerService(IConfiguration configuration,ILogger<ProducerService> logger, IOptions<KafkaOptionConfig> kafkaOptionConfig) public ProducerService(IConfiguration configuration,ILogger<ProducerService> logger, IOptions<KafkaOptionConfig> kafkaOptionConfig)
{ {
_configuration = configuration; _configuration = configuration;

2
services/JiShe.CollectBus.Application/CollectBusApplicationModule.cs
View File

@ -41,7 +41,7 @@ namespace JiShe.CollectBus;
typeof(CollectBusFreeRedisModule), typeof(CollectBusFreeRedisModule),
typeof(CollectBusFreeSqlModule), typeof(CollectBusFreeSqlModule),
typeof(CollectBusKafkaModule), typeof(CollectBusKafkaModule),
typeof(CollectBusIoTDBModule), typeof(CollectBusIoTDbModule),
typeof(CollectBusCassandraModule) typeof(CollectBusCassandraModule)
)] )]
public class CollectBusApplicationModule : AbpModule public class CollectBusApplicationModule : AbpModule

10
services/JiShe.CollectBus.Application/Samples/SampleAppService.cs
View File

@ -34,13 +34,13 @@ namespace JiShe.CollectBus.Samples;
public class SampleAppService : CollectBusAppService, ISampleAppService, IKafkaSubscribe public class SampleAppService : CollectBusAppService, ISampleAppService, IKafkaSubscribe
{ {
private readonly ILogger<SampleAppService> _logger; private readonly ILogger<SampleAppService> _logger;
private readonly IIoTDBProvider _iotDBProvider; private readonly IIoTDbProvider _iotDBProvider;
private readonly IoTDBRuntimeContext _dbContext; private readonly IoTDbRuntimeContext _dbContext;
private readonly IoTDBOptions _options; private readonly IoTDbOptions _options;
private readonly IRedisDataCacheService _redisDataCacheService; private readonly IRedisDataCacheService _redisDataCacheService;
public SampleAppService(IIoTDBProvider iotDBProvider, IOptions<IoTDBOptions> options, public SampleAppService(IIoTDbProvider iotDBProvider, IOptions<IoTDbOptions> options,
IoTDBRuntimeContext dbContext, ILogger<SampleAppService> logger, IRedisDataCacheService redisDataCacheService) IoTDbRuntimeContext dbContext, ILogger<SampleAppService> logger, IRedisDataCacheService redisDataCacheService)
{ {
_iotDBProvider = iotDBProvider; _iotDBProvider = iotDBProvider;
_options = options.Value; _options = options.Value;

4
services/JiShe.CollectBus.Application/ScheduledMeterReading/BasicScheduledMeterReadingService.cs
View File

@ -37,7 +37,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading
public abstract class BasicScheduledMeterReadingService : CollectBusAppService, IScheduledMeterReadingService public abstract class BasicScheduledMeterReadingService : CollectBusAppService, IScheduledMeterReadingService
{ {
private readonly ILogger<BasicScheduledMeterReadingService> _logger; private readonly ILogger<BasicScheduledMeterReadingService> _logger;
private readonly IIoTDBProvider _dbProvider; private readonly IIoTDbProvider _dbProvider;
private readonly IMeterReadingRecordRepository _meterReadingRecordRepository; private readonly IMeterReadingRecordRepository _meterReadingRecordRepository;
private readonly IProducerService _producerService; private readonly IProducerService _producerService;
private readonly IRedisDataCacheService _redisDataCacheService; private readonly IRedisDataCacheService _redisDataCacheService;
@ -48,7 +48,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading
IMeterReadingRecordRepository meterReadingRecordRepository, IMeterReadingRecordRepository meterReadingRecordRepository,
IProducerService producerService, IProducerService producerService,
IRedisDataCacheService redisDataCacheService, IRedisDataCacheService redisDataCacheService,
IIoTDBProvider dbProvider, IIoTDbProvider dbProvider,
IOptions<KafkaOptionConfig> kafkaOptions) IOptions<KafkaOptionConfig> kafkaOptions)
{ {
_logger = logger; _logger = logger;

2
services/JiShe.CollectBus.Application/ScheduledMeterReading/EnergySystemScheduledMeterReadingService.cs
View File

@ -37,7 +37,7 @@ namespace JiShe.CollectBus.ScheduledMeterReading
string serverTagName = string.Empty; string serverTagName = string.Empty;
public EnergySystemScheduledMeterReadingService( public EnergySystemScheduledMeterReadingService(
ILogger<EnergySystemScheduledMeterReadingService> logger, ILogger<EnergySystemScheduledMeterReadingService> logger,
IIoTDBProvider dbProvider, IIoTDbProvider dbProvider,
IMeterReadingRecordRepository meterReadingRecordRepository, IMeterReadingRecordRepository meterReadingRecordRepository,
IOptions<KafkaOptionConfig> kafkaOptions, IOptions<KafkaOptionConfig> kafkaOptions,
IProducerService producerService, IProducerService producerService,

4
services/JiShe.CollectBus.Application/Subscribers/SubscriberAppService.cs
View File

@ -31,7 +31,7 @@ namespace JiShe.CollectBus.Subscribers
private readonly IRepository<MessageReceivedLogin, Guid> _messageReceivedLoginEventRepository; private readonly IRepository<MessageReceivedLogin, Guid> _messageReceivedLoginEventRepository;
private readonly IRepository<MessageReceivedHeartbeat, Guid> _messageReceivedHeartbeatEventRepository; private readonly IRepository<MessageReceivedHeartbeat, Guid> _messageReceivedHeartbeatEventRepository;
private readonly IMeterReadingRecordRepository _meterReadingRecordsRepository; private readonly IMeterReadingRecordRepository _meterReadingRecordsRepository;
private readonly IIoTDBProvider _dbProvider; private readonly IIoTDbProvider _dbProvider;
/// <summary> /// <summary>
/// Initializes a new instance of the <see cref="SubscriberAppService"/> class. /// Initializes a new instance of the <see cref="SubscriberAppService"/> class.
@ -47,7 +47,7 @@ namespace JiShe.CollectBus.Subscribers
IServiceProvider serviceProvider, IServiceProvider serviceProvider,
IRepository<MessageReceivedLogin, Guid> messageReceivedLoginEventRepository, IRepository<MessageReceivedLogin, Guid> messageReceivedLoginEventRepository,
IRepository<MessageReceivedHeartbeat, Guid> messageReceivedHeartbeatEventRepository, IRepository<MessageReceivedHeartbeat, Guid> messageReceivedHeartbeatEventRepository,
IIoTDBProvider dbProvider, IIoTDbProvider dbProvider,
IMeterReadingRecordRepository meterReadingRecordsRepository) IMeterReadingRecordRepository meterReadingRecordsRepository)
{ {
_logger = logger; _logger = logger;