合并代码

modules/JiShe.CollectBus.Cassandra/CassandraQueryOptimizer.cs

@@ -1,156 +0,0 @@
-using System.Collections.Concurrent;
-using Cassandra;
-using Cassandra.Mapping;
-using Microsoft.Extensions.Caching.Memory;
-using Microsoft.Extensions.Logging;
-
-namespace JiShe.CollectBus.Cassandra
-{
-    public class CassandraQueryOptimizer
-    {
-        private readonly ISession _session;
-        private readonly ILogger<CassandraQueryOptimizer> _logger;
-        private readonly IMemoryCache _cache;
-        private readonly ConcurrentDictionary<string, PreparedStatement> _preparedStatements;
-        private readonly int _batchSize;
-        private readonly TimeSpan _cacheExpiration;
-
-        public CassandraQueryOptimizer(
-            ISession session,
-            ILogger<CassandraQueryOptimizer> logger,
-            IMemoryCache cache,
-            int batchSize = 100,
-            TimeSpan? cacheExpiration = null)
-        {
-            _session = session;
-            _logger = logger;
-            _cache = cache;
-            _preparedStatements = new ConcurrentDictionary<string, PreparedStatement>();
-            _batchSize = batchSize;
-            _cacheExpiration = cacheExpiration ?? TimeSpan.FromMinutes(5);
-        }
-
-        public async Task<PreparedStatement> GetOrPrepareStatementAsync(string cql)
-        {
-            return _preparedStatements.GetOrAdd(cql, key =>
-            {
-                try
-                {
-                    var statement = _session.Prepare(key);
-                    _logger.LogDebug($"Prepared statement for CQL: {key}");
-                    return statement;
-                }
-                catch (Exception ex)
-                {
-                    _logger.LogError(ex, $"Failed to prepare statement for CQL: {key}");
-                    throw;
-                }
-            });
-        }
-
-        public async Task ExecuteBatchAsync(IEnumerable<BoundStatement> statements)
-        {
-            var batch = new BatchStatement();
-            var count = 0;
-
-            foreach (var statement in statements)
-            {
-                batch.Add(statement);
-                count++;
-
-                if (count >= _batchSize)
-                {
-                    await ExecuteBatchAsync(batch);
-                    batch = new BatchStatement();
-                    count = 0;
-                }
-            }
-
-            if (count > 0)
-            {
-                await ExecuteBatchAsync(batch);
-            }
-        }
-
-        private async Task ExecuteBatchAsync(BatchStatement batch)
-        {
-            try
-            {
-                await _session.ExecuteAsync(batch);
-            }
-            catch (Exception ex)
-            {
-                _logger.LogError(ex, "Failed to execute batch statement");
-                throw;
-            }
-        }
-
-        public async Task<T> GetOrSetFromCacheAsync<T>(string cacheKey, Func<Task<T>> getData)
-        {
-            if (_cache.TryGetValue(cacheKey, out T cachedValue))
-            {
-                _logger.LogDebug($"Cache hit for key: {cacheKey}");
-                return cachedValue;
-            }
-
-            var data = await getData();
-            _cache.Set(cacheKey, data, _cacheExpiration);
-            _logger.LogDebug($"Cache miss for key: {cacheKey}, data cached");
-            return data;
-        }
-
-        public async Task<IEnumerable<T>> ExecutePagedQueryAsync<T>(
-            string cql,
-            object[] parameters,
-            int pageSize = 100,
-            string pagingState = null) where T : class 
-        {
-            var statement = await GetOrPrepareStatementAsync(cql);
-            var boundStatement = statement.Bind(parameters);
-
-            if (!string.IsNullOrEmpty(pagingState))
-            {
-                boundStatement.SetPagingState(Convert.FromBase64String(pagingState));
-            }
-
-            boundStatement.SetPageSize(pageSize);
-
-            try
-            {
-                var result = await _session.ExecuteAsync(boundStatement);
-                //TODO: RETURN OBJECT
-                throw new NotImplementedException();
-                //result.GetRows()
-                //return result.Select(row => row);
-            }
-            catch (Exception ex)
-            {
-                _logger.LogError(ex, $"Failed to execute paged query: {cql}");
-                throw;
-            }
-        }
-
-        public async Task BulkInsertAsync<T>(IEnumerable<T> items, string tableName)
-        {
-            var mapper = new Mapper(_session);
-            var batch = new List<BoundStatement>();
-            var cql = $"INSERT INTO {tableName} ({{0}}) VALUES ({{1}})";
-
-            foreach (var chunk in items.Chunk(_batchSize))
-            {
-                var statements = chunk.Select(item =>
-                {
-                    var props = typeof(T).GetProperties();
-                    var columns = string.Join(", ", props.Select(p => p.Name));
-                    var values = string.Join(", ", props.Select(p => "?"));
-                    var statement = _session.Prepare(string.Format(cql, columns, values));
-                    return statement.Bind(props.Select(p => p.GetValue(item)).ToArray());
-                });
-
-                batch.AddRange(statements);
-            }
-
-            await ExecuteBatchAsync(batch);
-        }
-    }
-} 

modules/JiShe.CollectBus.Cassandra/CassandraRepository.cs

@@ -1,19 +1,13 @@
-using Cassandra;
+using System.Linq.Expressions;
-using Cassandra.Data.Linq;
 using Cassandra.Mapping;
 using JiShe.CollectBus.Cassandra.Extensions;
-using JiShe.CollectBus.Common.Attributes;
-using Microsoft.AspNetCore.Http;
-using System.Reflection;
-using Thrift.Protocol.Entities;
 using Volo.Abp.Domain.Entities;
-using Volo.Abp.Domain.Repositories;
 
 namespace JiShe.CollectBus.Cassandra
 {
     public class CassandraRepository<TEntity, TKey>
         : ICassandraRepository<TEntity, TKey>
-        where TEntity : class
+        where TEntity : class, ICassandraEntity<TKey>
     {
         private readonly ICassandraProvider _cassandraProvider;
         public CassandraRepository(ICassandraProvider cassandraProvider, MappingConfiguration mappingConfig)
@@ -27,12 +21,29 @@ namespace JiShe.CollectBus.Cassandra
 
         public virtual async Task<TEntity> GetAsync(TKey id)
         {
-            return await Mapper.SingleOrDefaultAsync<TEntity>("WHERE id = ?", id);
+            return await GetAsync("WHERE id = ?", id);
         }
 
-        public virtual async Task<List<TEntity>> GetListAsync()
+        public virtual async Task<TEntity?> GetAsync(string cql, params object[] args)
         {
-            return (await Mapper.FetchAsync<TEntity>()).ToList();
+            return await Mapper.SingleAsync<TEntity?>(cql, args);
+        }
+
+
+        public virtual async Task<TEntity> FirstOrDefaultAsync(TKey id)
+        {
+            return await FirstOrDefaultAsync("WHERE id = ?", id);
+        }
+
+        public virtual async Task<TEntity?> FirstOrDefaultAsync(string cql, params object[] args)
+        {
+            return await Mapper.FirstOrDefaultAsync<TEntity>(cql, args);
+        }
+
+
+        public virtual async Task<List<TEntity>?> GetListAsync(string? cql = null, params object[] args)
+        {
+            return cql.IsNullOrWhiteSpace() ? (await Mapper.FetchAsync<TEntity>()).ToList() : (await Mapper.FetchAsync<TEntity>(cql, args)).ToList();
         }
 
         public virtual async Task<TEntity> InsertAsync(TEntity entity)

modules/JiShe.CollectBus.Cassandra/CollectBusCassandraModule.cs

@@ -1,7 +1,4 @@
-using Cassandra;
+using Microsoft.Extensions.DependencyInjection;
-using Cassandra.Mapping;
-using JiShe.CollectBus.Cassandra.Mappers;
-using Microsoft.Extensions.DependencyInjection;
 using Volo.Abp;
 using Volo.Abp.Autofac;
 using Volo.Abp.Modularity;

modules/JiShe.CollectBus.Cassandra/Extensions/ServiceCollectionExtensions.cs

@@ -1,10 +1,4 @@
-using Autofac.Core;
+using JiShe.CollectBus.Cassandra;
-using Cassandra;
-using Cassandra.Mapping;
-using JiShe.CollectBus.Cassandra;
-using JiShe.CollectBus.Cassandra.Mappers;
-using Microsoft.Extensions.Options;
-using System.Reflection;
 using Volo.Abp;
 using Volo.Abp.Modularity;
 
@@ -26,8 +20,6 @@ namespace Microsoft.Extensions.DependencyInjection
         public static void AddCassandra(this ServiceConfigurationContext context)
         {
             context.Services.AddTransient(typeof(ICassandraRepository<,>), typeof(CassandraRepository<,>));
-            context.Services.AddSingleton(new MappingConfiguration()
-                .Define(new CollectBusMapping()));
         }
     }
 }

modules/JiShe.CollectBus.Cassandra/Extensions/SessionExtension.cs

@@ -3,9 +3,7 @@ using System.Text;
 using Cassandra;
 using System.ComponentModel.DataAnnotations;
 using JiShe.CollectBus.Common.Attributes;
-using Cassandra.Mapping;
+using Volo.Abp.Data;
-using Cassandra.Data.Linq;
-using Thrift.Protocol.Entities;
 
 namespace JiShe.CollectBus.Cassandra.Extensions
 {
@@ -16,17 +14,26 @@ namespace JiShe.CollectBus.Cassandra.Extensions
             var type = typeof(TEntity);
             var tableAttribute = type.GetCustomAttribute<CassandraTableAttribute>();
             var tableName = tableAttribute?.Name ?? type.Name.ToLower();
-            //var tableKeyspace = tableAttribute?.Keyspace ?? defaultKeyspace;
             var tableKeyspace = session.Keyspace;
 
             var properties = type.GetProperties();
-            var primaryKey = properties.FirstOrDefault(p => p.GetCustomAttribute<KeyAttribute>() != null);
+
+            // 分区键设置
+            var primaryKey = properties.FirstOrDefault(p => p.GetCustomAttribute<PartitionKeyAttribute>() != null);
 
             if (primaryKey == null)
             {
                 throw new InvalidOperationException($"No primary key defined for type {type.Name}");
             }
 
+            // 集群键设置
+            var clusteringKeys = properties.Where(p => p.GetCustomAttribute<ClusteringKeyAttribute>() != null).Select(a=>a.Name).ToList();
+            var clusteringKeyCql = string.Empty;
+            if (clusteringKeys.Any())
+            {
+                clusteringKeyCql = $", {string.Join(", ", clusteringKeys)}";
+            }
+
             var cql = new StringBuilder();
             cql.Append($"CREATE TABLE IF NOT EXISTS {tableKeyspace}.{tableName} (");
 
@@ -40,7 +47,7 @@ namespace JiShe.CollectBus.Cassandra.Extensions
                 cql.Append($"{columnName} {cqlType}, ");
             }
             cql.Length -= 2; // Remove last comma and space
-            cql.Append($", PRIMARY KEY ({primaryKey.Name.ToLower()}))");
+            cql.Append($", PRIMARY KEY (({primaryKey.Name.ToLower()}){clusteringKeyCql}))");
 
             session.Execute(cql.ToString());
         }
@@ -61,6 +68,7 @@ namespace JiShe.CollectBus.Cassandra.Extensions
             if (type == typeof(Guid)) return "uuid";
             if (type == typeof(DateTimeOffset)) return "timestamp";
             if (type == typeof(Byte[])) return "blob";
+            if (type == typeof(ExtraPropertyDictionary)) return "map<text,text>";
 
             // 处理集合类型
             if (type.IsGenericType)
@@ -72,6 +80,8 @@ namespace JiShe.CollectBus.Cassandra.Extensions
                     return $"list<{GetCassandraType(elementType)}>";
                 if (genericType == typeof(HashSet<>))
                     return $"set<{GetCassandraType(elementType)}>";
+                if (genericType == typeof(Nullable<>))
+                    return GetCassandraType(elementType);
                 if (genericType == typeof(Dictionary<,>))
                 {
                     var keyType = type.GetGenericArguments()[0];

modules/JiShe.CollectBus.Cassandra/ICassandraRepository.cs

@@ -10,7 +10,10 @@ namespace JiShe.CollectBus.Cassandra
     public interface ICassandraRepository<TEntity, TKey> where TEntity : class
     {
         Task<TEntity> GetAsync(TKey id);
-        Task<List<TEntity>> GetListAsync();
+        Task<TEntity?> GetAsync(string cql, params object[] args);
+        Task<TEntity> FirstOrDefaultAsync(TKey id);
+        Task<TEntity?> FirstOrDefaultAsync(string cql, params object[] args);
+        Task<List<TEntity>?> GetListAsync(string? cql = null, params object[] args);
         Task<TEntity> InsertAsync(TEntity entity);
         Task<TEntity> UpdateAsync(TEntity entity);
         Task DeleteAsync(TEntity entity);

modules/JiShe.CollectBus.Cassandra/JiShe.CollectBus.Cassandra.csproj

@@ -15,8 +15,8 @@
   </ItemGroup>
 
   <ItemGroup>
-    <ProjectReference Include="..\..\services\JiShe.CollectBus.Domain\JiShe.CollectBus.Domain.csproj" />
     <ProjectReference Include="..\..\shared\JiShe.CollectBus.Common\JiShe.CollectBus.Common.csproj" />
+    <ProjectReference Include="..\..\shared\JiShe.CollectBus.Domain.Shared\JiShe.CollectBus.Domain.Shared.csproj" />
   </ItemGroup>
 
 </Project>

modules/JiShe.CollectBus.IoTDB/CollectBusIoTDBModule.cs

@@ -1,33 +1,22 @@
 using JiShe.CollectBus.IoTDB.Context;
-using JiShe.CollectBus.IoTDB.Interface;
 using JiShe.CollectBus.IoTDB.Options;
-using JiShe.CollectBus.IoTDB.Provider;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
 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)
     {
-             
         var configuration = context.Services.GetConfiguration();
-            Configure<IoTDBOptions>(options =>
+        Configure<IoTDbOptions>(options => { configuration.GetSection(nameof(IoTDbOptions)).Bind(options); });
-            {
-                configuration.GetSection(nameof(IoTDBOptions)).Bind(options);
-            });
 
         // 注册上下文为Scoped
-            context.Services.AddScoped<IoTDBRuntimeContext>();
+        context.Services.AddScoped<IoTDbRuntimeContext>();
-
-            // 注册Session工厂
-            context.Services.AddSingleton<IIoTDBSessionFactory, IoTDBSessionFactory>();
-
-            // 注册Provider
-            context.Services.AddScoped<IIoTDBProvider, IoTDBProvider>();
-
-        }
     }
 }

modules/JiShe.CollectBus.IoTDB/Context/IoTDBRuntimeContext.cs

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

modules/JiShe.CollectBus.IoTDB/Interface/IIoTDBProvider.cs

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

modules/JiShe.CollectBus.IoTDB/Interface/IIoTDBSessionFactory.cs

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

modules/JiShe.CollectBus.IoTDB/Interface/IIoTDBSessionPool.cs

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

modules/JiShe.CollectBus.IoTDB/Options/IoTDBOptions.cs

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

modules/JiShe.CollectBus.IoTDB/Provider/IoTDBProvider.cs

@@ -17,20 +17,26 @@ namespace JiShe.CollectBus.IoTDB.Provider
     /// <summary>
     /// IoTDB数据源
     /// </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);
 
-        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;
             _sessionFactory = sessionFactory;

modules/JiShe.CollectBus.IoTDB/Provider/IoTDBSessionFactory.cs

@@ -2,6 +2,7 @@
 using JiShe.CollectBus.IoTDB.Interface;
 using JiShe.CollectBus.IoTDB.Options;
 using Microsoft.Extensions.Options;
+using Volo.Abp.DependencyInjection;
 
 namespace JiShe.CollectBus.IoTDB.Provider
 {
@@ -9,25 +10,29 @@ namespace JiShe.CollectBus.IoTDB.Provider
     /// <summary>
     /// 实现带缓存的Session工厂
     /// </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;
 
-        public IoTDBSessionFactory(IOptions<IoTDBOptions> options)
+        /// <summary>
+        /// IoTDbSessionFactory
+        /// </summary>
+        /// <param name="options"></param>
+        public IoTDbSessionFactory(IOptions<IoTDbOptions> options)
         {
             _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 =>
             {
                 var pool = key
-                    ? (IIoTDBSessionPool)new TableSessionPoolAdapter(_options)
+                    ? (IIoTDbSessionPool)new TableSessionPoolAdapter(_options)
                     : new SessionPoolAdapter(_options);
 
                 pool.OpenAsync().ConfigureAwait(false).GetAwaiter().GetResult(); ;

modules/JiShe.CollectBus.IoTDB/Provider/SessionPoolAdapter.cs

@@ -9,12 +9,16 @@ namespace JiShe.CollectBus.IoTDB.Provider
     /// <summary>
     /// 树模型连接池
     /// </summary>
-    public class SessionPoolAdapter : IIoTDBSessionPool
+    public class SessionPoolAdapter : IIoTDbSessionPool
     {
         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;
             _sessionPool = new SessionPool.Builder()

modules/JiShe.CollectBus.IoTDB/Provider/TableSessionPoolAdapter.cs

@@ -9,12 +9,16 @@ namespace JiShe.CollectBus.IoTDB.Provider
     /// <summary>
     /// 表模型Session连接池
     /// </summary>
-    public class TableSessionPoolAdapter : IIoTDBSessionPool
+    public class TableSessionPoolAdapter : IIoTDbSessionPool
     {
         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;
             _sessionPool = new TableSessionPool.Builder()

modules/JiShe.CollectBus.Kafka/AdminClient/AdminClientService.cs

@@ -1,30 +1,24 @@
 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 Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.Logging;
 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.
+    /// Initializes a new instance of the <see cref="AdminClientService" /> class.
     /// </summary>
-        /// <param name="configuration">The configuration.</param>
+    /// <param name="configuration"></param>
-        /// <param name="logger">The logger.</param>
+    /// <param name="logger"></param>
     public AdminClientService(IConfiguration configuration, ILogger<AdminClientService> logger)
     {
         _logger = logger;
-            GetInstance(configuration);
+        Instance = GetInstance(configuration);
     }
 
     /// <summary>
@@ -33,70 +27,17 @@ namespace JiShe.CollectBus.Kafka.AdminClient
     /// <value>
     ///     The instance.
     /// </value>
-        public IAdminClient Instance { get; set; } = default;
+    public IAdminClient Instance { get; set; }
 
     /// <summary>
-        /// 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"],
-            };
-            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="topic"></param>
-        /// <param name="numPartitions">主题分区数量</param>
+    /// <param name="numPartitions"></param>
-        /// <param name="replicationFactor">副本数量，不能高于Brokers数量</param>
+    /// <param name="replicationFactor"></param>
     /// <returns></returns>
     public async Task CreateTopicAsync(string topic, int numPartitions, short replicationFactor)
     {
-
         try
         {
             if (await CheckTopicAsync(topic)) return;
@@ -114,10 +55,7 @@ namespace JiShe.CollectBus.Kafka.AdminClient
         }
         catch (CreateTopicsException e)
         {
-                if (e.Results[0].Error.Code != ErrorCode.TopicAlreadyExists)
+            if (e.Results[0].Error.Code != ErrorCode.TopicAlreadyExists) throw;
-                {
-                    throw;
-                }
         }
     }
 
@@ -200,5 +138,53 @@ namespace JiShe.CollectBus.Kafka.AdminClient
     {
         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));
     }
 }

modules/JiShe.CollectBus.Kafka/Attributes/KafkaSubscribeAttribute.cs

@@ -1,14 +1,26 @@
-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)
     {
+        Topic = topic;
+    }
+
+    /// <summary>
+    ///     订阅主题
+    /// </summary>
+    public KafkaSubscribeAttribute(string topic, int partition)
+    {
+        Topic = topic;
+        Partition = partition;
+    }
+
     /// <summary>
     ///     订阅的主题
     /// </summary>
@@ -22,7 +34,7 @@ namespace JiShe.CollectBus.Kafka.Attributes
     /// <summary>
     ///     消费者组
     /// </summary>
-        public string? GroupId { get; set; } = null;//"default"
+    public string? GroupId { get; set; } = null; //"default"
 
     /// <summary>
     ///     任务数(默认是多少个分区多少个任务)
@@ -44,25 +56,5 @@ namespace JiShe.CollectBus.Kafka.Attributes
     ///     批次超时时间
     ///     格式:("00:05:00")
     /// </summary>
-        public TimeSpan? BatchTimeout { get; set; }=null;
+    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;
-        }
-    }
 }

modules/JiShe.CollectBus.Kafka/Attributes/TopicAttribute.cs

@@ -1,16 +1,10 @@
-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)]
-    public class TopicAttribute: Attribute
-    {
     /// <summary>
-        /// Initializes a new instance of the <see cref="TopicAttribute"/> class.
+    ///     Initializes a new instance of the <see cref="TopicAttribute" /> class.
     /// </summary>
     /// <param name="name">The name.</param>
     public TopicAttribute(string name = "Default")
@@ -25,5 +19,4 @@ namespace JiShe.CollectBus.Kafka.Attributes
     ///     The name.
     /// </value>
     public string Name { get; set; }
-    }
 }

modules/JiShe.CollectBus.Kafka/Consumer/ConsumerService.cs

@@ -13,15 +13,18 @@ namespace JiShe.CollectBus.Kafka.Consumer
     public class ConsumerService : IConsumerService, IDisposable
     {
         private readonly ILogger<ConsumerService> _logger;
-        private readonly IConfiguration _configuration;
         private readonly ConcurrentDictionary<Type, (object Consumer, CancellationTokenSource CTS)>
             _consumerStore = new();
         private readonly KafkaOptionConfig _kafkaOptionConfig;
         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;
             _kafkaOptionConfig = kafkaOptionConfig.Value;
         }
@@ -165,10 +168,10 @@ namespace JiShe.CollectBus.Kafka.Consumer
         /// <summary>
         /// 订阅消息
         /// </summary>
-        /// <typeparam name="TKey"></typeparam>
         /// <typeparam name="TValue"></typeparam>
         /// <param name="topics"></param>
         /// <param name="messageHandler"></param>
+        /// <param name="groupId"></param>
         /// <returns></returns>
         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>
         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);
 
         }
 

modules/JiShe.CollectBus.Kafka/Consumer/IConsumerService.cs

@@ -1,15 +1,9 @@
-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>
     ///     订阅消息
@@ -18,9 +12,11 @@ namespace JiShe.CollectBus.Kafka.Consumer
     /// <param name="topic"></param>
     /// <param name="messageHandler"></param>
     /// <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>
@@ -31,16 +27,24 @@ namespace JiShe.CollectBus.Kafka.Consumer
     /// <param name="topics"></param>
     /// <param name="messageHandler"></param>
     /// <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;
-    }
 }

modules/JiShe.CollectBus.Kafka/Internal/HeadersFilter.cs

@@ -1,17 +1,12 @@
 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>
-    public class HeadersFilter : Dictionary<string, byte[]>
-    {
     /// <summary>
     ///     判断Headers是否匹配
     /// </summary>
@@ -20,11 +15,8 @@ namespace JiShe.CollectBus.Kafka.Internal
     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;
     }
-    }
 }

modules/JiShe.CollectBus.Kafka/Internal/IKafkaSubscribe.cs

@@ -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
-    {
-    }
 }

modules/JiShe.CollectBus.Kafka/Internal/ISubscribeAck.cs

@@ -1,13 +1,7 @@
-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>
@@ -17,5 +11,4 @@ namespace JiShe.CollectBus.Kafka.Internal
     ///     消息
     /// </summary>
     string? Msg { get; set; }
-    }
 }

modules/JiShe.CollectBus.Kafka/Internal/KafkaOptionConfig.cs

@@ -1,14 +1,9 @@
 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>
@@ -17,7 +12,7 @@ namespace JiShe.CollectBus.Kafka.Internal
     /// <summary>
     ///     服务器标识
     /// </summary>
-        public string ServerTagName { get; set; }= "KafkaFilterKey";
+    public string ServerTagName { get; set; } = "KafkaFilterKey";
 
     /// <summary>
     ///     kafka主题副本数量
@@ -32,7 +27,7 @@ namespace JiShe.CollectBus.Kafka.Internal
     /// <summary>
     ///     是否开启过滤器
     /// </summary>
-        public bool EnableFilter { get; set; }= true;
+    public bool EnableFilter { get; set; } = true;
 
     /// <summary>
     ///     是否开启认证
@@ -47,7 +42,7 @@ namespace JiShe.CollectBus.Kafka.Internal
     /// <summary>
     ///     认证方式
     /// </summary>
-        public SaslMechanism SaslMechanism { get; set; }= SaslMechanism.Plain;
+    public SaslMechanism SaslMechanism { get; set; } = SaslMechanism.Plain;
 
     /// <summary>
     ///     用户名
@@ -63,6 +58,4 @@ namespace JiShe.CollectBus.Kafka.Internal
     ///     首次采集时间
     /// </summary>
     public DateTime FirstCollectionTime { get; set; }
-
-    }
 }

modules/JiShe.CollectBus.Kafka/Internal/ReflectionHelper.cs

@@ -1,25 +1,19 @@
-using Newtonsoft.Json;
+using System.Collections;
-using System;
-using System.Collections;
-using System.Collections.Generic;
-using System.Linq;
 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>
-    /// 反射辅助类
+    ///     集合类型
+    ///     Item1：参数类型
+    ///     Item2：集合元素类型
     /// </summary>
-    public static class ReflectionHelper
+    public static Tuple<Type, Type?> GetParameterTypeInfo(this MethodInfo method, int parameterIndex = 0)
-    {
-        /// <summary>
-        ///集合类型
-        ///Item1：参数类型
-        ///Item2：集合元素类型
-        /// </summary>
-        public static Tuple<Type,Type?> GetParameterTypeInfo(this MethodInfo method, int parameterIndex=0)
     {
         // 参数校验
         if (method == null) throw new ArgumentNullException(nameof(method));
@@ -27,18 +21,15 @@ namespace JiShe.CollectBus.Kafka.Internal
         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;
 
         // 判断是否是集合类型（排除字符串）
         if (paramType != typeof(string) && IsEnumerableType(paramType))
-            {
             elementType = GetEnumerableElementType(paramType);
-            }
 
         return Tuple.Create(paramType, elementType);
-            
     }
 
     /// <summary>
@@ -50,7 +41,7 @@ namespace JiShe.CollectBus.Kafka.Internal
                || (type.IsGenericType && type.GetInterfaces()
                    .Any(t => t.IsGenericType
                              && t.GetGenericTypeDefinition() == typeof(IEnumerable<>)))
-                || type.GetInterfaces().Any(t => t == typeof(System.Collections.IEnumerable));
+               || type.GetInterfaces().Any(t => t == typeof(IEnumerable));
     }
 
     /// <summary>
@@ -81,20 +72,20 @@ namespace JiShe.CollectBus.Kafka.Internal
     }
 
 
-        // <summary>
+    /// <summary>
     /// 判断是否使用强转换
     /// </summary>
-        /// <param name="targetType">目标类型</param>
+    /// <param name="targetType"></param>
     /// <returns></returns>
     public static bool IsConvertType(this Type targetType)
     {
         // 处理可空类型
-            Type underlyingType = Nullable.GetUnderlyingType(targetType) ?? targetType;
+        var underlyingType = Nullable.GetUnderlyingType(targetType) ?? targetType;
         // 情况1：值类型或基元类型（如 int、DateTime）
         if (underlyingType.IsValueType || underlyingType.IsPrimitive)
             return true;
         // 情况2：字符串类型直接赋值
-            else if (underlyingType == typeof(string))
+        if (underlyingType == typeof(string))
             return true;
 
         // 情况3：枚举类型处理
@@ -109,5 +100,4 @@ namespace JiShe.CollectBus.Kafka.Internal
         //}
         return false;
     }
-    }
 }

modules/JiShe.CollectBus.Kafka/Internal/SubscribeResult.cs

@@ -1,15 +1,7 @@
-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>
@@ -35,9 +27,7 @@ namespace JiShe.CollectBus.Kafka.Internal
     /// <summary>
     /// 失败
     /// </summary>
-        /// <param name="code"></param>
     /// <param name="msg"></param>
-        /// <param name="data"></param>
     /// <returns></returns>
     public SubscribeResult Fail(string? msg = null)
     {
@@ -45,11 +35,10 @@ namespace JiShe.CollectBus.Kafka.Internal
         Ack = false;
         return this;
     }
-    }
+}
-
-    public static partial class SubscribeAck
-    {
 
+public static class SubscribeAck
+{
     /// <summary>
     ///     成功
     /// </summary>
@@ -70,6 +59,4 @@ namespace JiShe.CollectBus.Kafka.Internal
     {
         return new SubscribeResult().Fail(msg);
     }
-    }
-
 }

modules/JiShe.CollectBus.Kafka/KafkaSubscribeExtensions.cs

@@ -8,26 +8,17 @@ using JiShe.CollectBus.Kafka.Consumer;
 using JiShe.CollectBus.Kafka.Internal;
 using JiShe.CollectBus.Kafka.Serialization;
 using Microsoft.AspNetCore.Builder;
-using Microsoft.AspNetCore.Mvc.Abstractions;
-using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
-using System;
 using System.Collections;
-using System.Collections.Generic;
 using System.ComponentModel;
-using System.Linq.Expressions;
 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
 {
-    public static class KafkaSubcribesExtensions
+    public static class KafkaSubscribeExtensions
     {
 
         public static void UseInitKafkaTopic(this IServiceProvider provider)
@@ -36,7 +27,7 @@ namespace JiShe.CollectBus.Kafka
             var kafkaAdminClient = provider.GetRequiredService<IAdminClientService>();
             var kafkaOptions = provider.GetRequiredService<IOptions<KafkaOptionConfig>>();
 
-            List<string> topics = ProtocolConstExtensions.GetAllTopicNamesByIssued();
+            var topics = ProtocolConstExtensions.GetAllTopicNamesByIssued();
             topics.AddRange(ProtocolConstExtensions.GetAllTopicNamesByReceived());
 
             foreach (var item in topics)
@@ -48,8 +39,6 @@ namespace JiShe.CollectBus.Kafka
         /// <summary>
         /// 添加Kafka订阅
         /// </summary>
-        /// <param name="app"></param>
-        /// <param name="assembly"></param>
         public static void UseKafkaSubscribe(this IServiceProvider provider)
         {
             var lifetime = provider.GetRequiredService<IHostApplicationLifetime>();
@@ -57,8 +46,8 @@ namespace JiShe.CollectBus.Kafka
             lifetime.ApplicationStarted.Register(() =>
             {
                 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);
                 if (string.IsNullOrWhiteSpace(assemblyPath))
                 {
@@ -98,6 +87,9 @@ namespace JiShe.CollectBus.Kafka
             });
         }
 
+        /// <summary>
+        /// 添加Kafka订阅
+        /// </summary>
         public static void UseKafkaSubscribersAsync(this IApplicationBuilder app, Assembly assembly)
         {
             var provider = app.ApplicationServices;
@@ -134,8 +126,6 @@ namespace JiShe.CollectBus.Kafka
         /// <summary>
         /// 构建Kafka订阅
         /// </summary>
-        /// <param name="subscribe"></param>
-        /// <param name="provider"></param>
         private static Tuple<int, int> BuildKafkaSubscribe(object subscribe, IServiceProvider provider, ILogger<CollectBusKafkaModule> logger, KafkaOptionConfig kafkaOptionConfig)
         {
             var subscribedMethods = subscribe.GetType().GetMethods()
@@ -169,11 +159,6 @@ namespace JiShe.CollectBus.Kafka
         /// <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)
         {
             var consumerService = provider.GetRequiredService<IConsumerService>();
@@ -225,10 +210,6 @@ namespace JiShe.CollectBus.Kafka
         /// <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)
         {
             var parameters = method.GetParameters();
@@ -351,9 +332,6 @@ namespace JiShe.CollectBus.Kafka
             }
             return false;
         }
-
-
-
     }
 
 

modules/JiShe.CollectBus.Kafka/Producer/IProducerService.cs

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

modules/JiShe.CollectBus.Kafka/Producer/ProducerService.cs

@@ -23,6 +23,13 @@ namespace JiShe.CollectBus.Kafka.Producer
         private readonly ConcurrentDictionary<Type, object> _producerCache = new();
         private class KafkaProducer<TKey, TValue> where TKey : notnull where TValue : class { }
         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)
         {
             _configuration = configuration;

services/JiShe.CollectBus.Application.Contracts/Samples/SampleDto.cs

@@ -4,14 +4,12 @@ using Volo.Abp.EventBus;
 namespace JiShe.CollectBus.Samples;
 
 [EventName("Sample.Kafka.Test")]
-[TopicName("Test1")]
 public class SampleDto
 {
     public int Value { get; set; }
 }
 
 [EventName("Sample.Kafka.Test2")]
-[TopicName("Test2")]
 public class SampleDto2
 {
     public int Value { get; set; }

services/JiShe.CollectBus.Application/CollectBusApplicationModule.cs

@@ -11,9 +11,11 @@ using System.Collections.Generic;
 using System.Linq;
 using System.Reflection;
 using System.Threading.Tasks;
+using Cassandra.Mapping;
 using JiShe.CollectBus.Cassandra;
 using JiShe.CollectBus.FreeRedis;
 using JiShe.CollectBus.IoTDB;
+using JiShe.CollectBus.Mappers;
 using Volo.Abp;
 using Volo.Abp.Application;
 using Volo.Abp.Autofac;
@@ -24,6 +26,8 @@ using Volo.Abp.EventBus;
 using Volo.Abp.Modularity;
 using Microsoft.Extensions.Options;
 using JiShe.CollectBus.Kafka.Internal;
+using JiShe.CollectBus.Interceptors;
+using JiShe.CollectBus.Common.Attributes;
 
 namespace JiShe.CollectBus;
 
@@ -37,7 +41,7 @@ namespace JiShe.CollectBus;
     typeof(CollectBusFreeRedisModule),
     typeof(CollectBusFreeSqlModule),
     typeof(CollectBusKafkaModule),
-    typeof(CollectBusIoTDBModule),
+    typeof(CollectBusIoTDbModule),
     typeof(CollectBusCassandraModule)
     )]
 public class CollectBusApplicationModule : AbpModule
@@ -51,6 +55,23 @@ public class CollectBusApplicationModule : AbpModule
         {
             options.AddMaps<CollectBusApplicationModule>(validate: true);
         });
+
+        context.Services.AddSingleton(new MappingConfiguration()
+            .Define(new CollectBusMapping()));
+
+        // 注册拦截器
+        context.Services.OnRegistered(ctx =>
+        {
+            var methods = ctx.ImplementationType.GetMethods();
+            foreach (var method in methods)
+            {
+                var attr = method.GetCustomAttribute(typeof(LogInterceptAttribute), true);
+                if (attr != null)
+                {
+                    ctx.Interceptors.TryAdd<LogInterceptor>();
+                }
+            }
+        });
     }
 
     public override async Task OnApplicationInitializationAsync(
@@ -70,10 +91,9 @@ public class CollectBusApplicationModule : AbpModule
 
         //初始化主题信息
         var kafkaAdminClient = context.ServiceProvider.GetRequiredService<IAdminClientService>();
-        var configuration = context.ServiceProvider.GetRequiredService<IConfiguration>();
         var kafkaOptions = context.ServiceProvider.GetRequiredService<IOptions<KafkaOptionConfig>>();
 
-        List<string> topics = ProtocolConstExtensions.GetAllTopicNamesByIssued();
+        var topics = ProtocolConstExtensions.GetAllTopicNamesByIssued();
         topics.AddRange(ProtocolConstExtensions.GetAllTopicNamesByReceived());
 
         foreach (var item in topics)

services/JiShe.CollectBus.Application/Interceptors/LogInterceptAttribute.cs

@@ -0,0 +1,9 @@
+using System;
+
+namespace JiShe.CollectBus.Interceptors
+{
+    [AttributeUsage(AttributeTargets.Method)]
+    public class LogInterceptAttribute : Attribute
+    {
+    }
+}

services/JiShe.CollectBus.Application/Interceptors/LogInterceptor.cs

@@ -0,0 +1,60 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using Volo.Abp.DependencyInjection;
+using Volo.Abp.DynamicProxy;
+
+namespace JiShe.CollectBus.Interceptors
+{
+    public class LogInterceptor : AbpInterceptor, ITransientDependency
+    {
+        public override async Task InterceptAsync(IAbpMethodInvocation invocation)
+        {
+            // 方法执行前的逻辑（如果需要）
+
+            try
+            {
+                // 执行原始方法
+                await invocation.ProceedAsync();
+
+                // 方法执行成功后，返回前的逻辑
+                await OnSuccessAsync(invocation);
+            }
+            catch (Exception ex)
+            {
+                // 出现异常时的逻辑
+                await OnExceptionAsync(invocation, ex);
+                throw;
+            }
+            finally
+            {
+                // 方法结束前一定会执行的逻辑
+                await OnCompleteAsync(invocation);
+            }
+        }
+
+        private Task OnSuccessAsync(IAbpMethodInvocation invocation)
+        {
+            // 方法执行成功后的逻辑
+            // 可以访问 invocation.ReturnValue 获取返回值
+            Console.WriteLine($"方法 {invocation.Method.Name} 成功执行，返回值：{invocation.ReturnValue}");
+            return Task.CompletedTask;
+        }
+
+        private Task OnExceptionAsync(IAbpMethodInvocation invocation, Exception ex)
+        {
+            // 方法执行异常时的逻辑
+            Console.WriteLine($"方法 {invocation.Method.Name} 执行异常：{ex.Message}");
+            return Task.CompletedTask;
+        }
+
+        private Task OnCompleteAsync(IAbpMethodInvocation invocation)
+        {
+            // 无论成功失败，方法执行完毕后的逻辑
+            Console.WriteLine($"方法 {invocation.Method.Name} 执行完毕");
+            return Task.CompletedTask;
+        }
+    }
+}

services/JiShe.CollectBus.Application/Mappers/CollectBusMapping.cs

@@ -1,13 +1,8 @@
 using Cassandra.Mapping;
-using System;
+using JiShe.CollectBus.IotSystems.Devices;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
 using JiShe.CollectBus.IotSystems.MessageIssueds;
-using static Cassandra.QueryTrace;
 
-namespace JiShe.CollectBus.Cassandra.Mappers
+namespace JiShe.CollectBus.Mappers
 {
     public class CollectBusMapping: Mappings
     {
@@ -15,6 +10,8 @@ namespace JiShe.CollectBus.Cassandra.Mappers
         {
             For<MessageIssued>()
                 .Column(e => e.Type, cm => cm.WithName("type").WithDbType<int>());
+            For<Device>()
+                .Column(e => e.Status, cm => cm.WithName("status").WithDbType<int>());
         }
     }
 }

services/JiShe.CollectBus.Application/Samples/SampleAppService.cs

@@ -34,13 +34,13 @@ namespace JiShe.CollectBus.Samples;
 public class SampleAppService : CollectBusAppService, ISampleAppService, IKafkaSubscribe
 {
     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;
 
-    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;
         _options = options.Value;

services/JiShe.CollectBus.Application/Samples/TestAppService.cs

@@ -25,6 +25,7 @@ using Volo.Abp.Domain.Repositories;
 using System.Diagnostics;
 using System.Linq;
 using Cassandra;
+using JiShe.CollectBus.Interceptors;
 
 namespace JiShe.CollectBus.Samples;
 
@@ -35,11 +36,10 @@ public class TestAppService : CollectBusAppService
     private readonly ICassandraRepository<MessageIssued, string> _messageReceivedCassandraRepository;
     private readonly ICassandraProvider _cassandraProvider;
 
-
-
     public TestAppService(
         ILogger<TestAppService> logger,
-        ICassandraRepository<MessageIssued, string> messageReceivedCassandraRepository, ICassandraProvider cassandraProvider)
+        ICassandraRepository<MessageIssued, string> messageReceivedCassandraRepository,
+        ICassandraProvider cassandraProvider)
     {
         _logger = logger;
         _messageReceivedCassandraRepository = messageReceivedCassandraRepository;
@@ -122,4 +122,11 @@ public class TestAppService : CollectBusAppService
         // 等待所有批处理完成
         await Task.WhenAll(tasks);
     }
+
+    [LogIntercept]
+    public async Task<string> LogInterceptorTest(string str)
+    {
+        _logger.LogWarning(str);
+        return str;
+    }
 }

services/JiShe.CollectBus.Application/ScheduledMeterReading/BasicScheduledMeterReadingService.cs

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

services/JiShe.CollectBus.Application/ScheduledMeterReading/EnergySystemScheduledMeterReadingService.cs

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

services/JiShe.CollectBus.Application/Subscribers/SubscriberAppService.cs

@@ -30,10 +30,8 @@ namespace JiShe.CollectBus.Subscribers
         private readonly IServiceProvider _serviceProvider;
         private readonly IRepository<MessageReceivedLogin, Guid> _messageReceivedLoginEventRepository;
         private readonly IRepository<MessageReceivedHeartbeat, Guid> _messageReceivedHeartbeatEventRepository;
-        private readonly IRepository<MessageReceived, Guid> _messageReceivedEventRepository;
-        private readonly IRepository<Device, Guid> _deviceRepository;
         private readonly IMeterReadingRecordRepository _meterReadingRecordsRepository;
-        private readonly IIoTDBProvider _dbProvider;
+        private readonly IIoTDbProvider _dbProvider;
 
         /// <summary>
         /// Initializes a new instance of the <see cref="SubscriberAppService"/> class.
@@ -43,16 +41,13 @@ namespace JiShe.CollectBus.Subscribers
         /// <param name="serviceProvider">The service provider.</param>
         /// <param name="messageReceivedLoginEventRepository">The message received login event repository.</param>
         /// <param name="messageReceivedHeartbeatEventRepository">The message received heartbeat event repository.</param>
-        /// <param name="messageReceivedEventRepository">The message received event repository.</param>
-        /// <param name="deviceRepository">The device repository.</param>
         /// <param name="meterReadingRecordsRepository">The device repository.</param>
         public SubscriberAppService(ILogger<SubscriberAppService> logger,
-            ITcpService tcpService, IServiceProvider serviceProvider,
+            ITcpService tcpService,
+            IServiceProvider serviceProvider,
             IRepository<MessageReceivedLogin, Guid> messageReceivedLoginEventRepository,
             IRepository<MessageReceivedHeartbeat, Guid> messageReceivedHeartbeatEventRepository,
-            IRepository<MessageReceived, Guid> messageReceivedEventRepository,
+            IIoTDbProvider dbProvider,
-            IRepository<Device, Guid> deviceRepository,
-            IIoTDBProvider dbProvider,
             IMeterReadingRecordRepository meterReadingRecordsRepository)
         {
             _logger = logger;
@@ -60,8 +55,6 @@ namespace JiShe.CollectBus.Subscribers
             _serviceProvider = serviceProvider;
             _messageReceivedLoginEventRepository = messageReceivedLoginEventRepository;
             _messageReceivedHeartbeatEventRepository = messageReceivedHeartbeatEventRepository;
-            _messageReceivedEventRepository = messageReceivedEventRepository;
-            _deviceRepository = deviceRepository;
             _meterReadingRecordsRepository = meterReadingRecordsRepository;
             _dbProvider = dbProvider;
         }

services/JiShe.CollectBus.Application/Workers/EpiCollectWorker.cs

@@ -10,7 +10,6 @@ using Volo.Abp.Uow;
 
 namespace JiShe.CollectBus.Workers
 {
-    [IgnoreJob]
     public class EpiCollectWorker : HangfireBackgroundWorkerBase, ITransientDependency,ICollectWorker
     {
         private readonly ILogger<EpiCollectWorker> _logger;

services/JiShe.CollectBus.Domain/IotSystems/Devices/Device.cs

@@ -1,14 +1,18 @@
 using System;
 using System.Collections.Generic;
+using System.ComponentModel.DataAnnotations;
 using System.Linq;
 using System.Text;
 using System.Threading.Tasks;
+using JiShe.CollectBus.Common.Attributes;
 using JiShe.CollectBus.Enums;
+using Volo.Abp.Auditing;
 using Volo.Abp.Domain.Entities;
+using Volo.Abp.Logging;
 
 namespace JiShe.CollectBus.IotSystems.Devices
 {
-    public class Device : AggregateRoot<Guid>
+    public class Device : BasicAggregateRoot<Guid>
     {
         /// <summary>
         /// Device
@@ -20,6 +24,7 @@ namespace JiShe.CollectBus.IotSystems.Devices
         /// <param name="status"></param>
         public Device(string number, string clientId, DateTime firstOnlineTime, DateTime lastOnlineTime, DeviceStatus status)
         {
+            Id = Guid.NewGuid();
             Number = number;
             FirstOnlineTime = firstOnlineTime;
             LastOnlineTime = lastOnlineTime;
@@ -30,6 +35,7 @@ namespace JiShe.CollectBus.IotSystems.Devices
         /// <summary>
         /// 集中器编号，在集中器登录时解析获取，并会更新为当前TCP连接的最新ClientId
         /// </summary>
+        [PartitionKey]
         public string Number { get; set; }
 
         /// <summary>
@@ -55,6 +61,7 @@ namespace JiShe.CollectBus.IotSystems.Devices
         /// <summary>
         /// 设备状态
         /// </summary>
+        [PartitionKey]
         public DeviceStatus Status { get; set; }
 
         /// <summary>

services/JiShe.CollectBus.Domain/IotSystems/MessageIssueds/MessageIssued.cs

@@ -11,19 +11,14 @@ using Volo.Abp.Domain.Entities;
 namespace JiShe.CollectBus.IotSystems.MessageIssueds
 {
     [CassandraTable]
-    public class MessageIssued:IEntity<string>
+    public class MessageIssued: ICassandraEntity<string>
     {
         public string ClientId { get; set; }
         public byte[] Message { get; set; }
         public string DeviceNo { get; set; }
         public IssuedEventType Type { get; set; }
         public string MessageId { get; set; }
-        [Key]
+        [PartitionKey]
         public string Id { get; set; }
-
-        public object?[] GetKeys()
-        {
-            return new object[] { Id };
-        }
     }
 }

shared/JiShe.CollectBus.Common/Attributes/CassandraTableAttribute.cs

@@ -21,4 +21,20 @@ namespace JiShe.CollectBus.Common.Attributes
         {
         }
     }
+
+    [AttributeUsage(AttributeTargets.Property | AttributeTargets.Field)]
+    public class PartitionKeyAttribute : Attribute
+    {
+        public PartitionKeyAttribute()
+        {
+        }
+    }
+
+    [AttributeUsage(AttributeTargets.Property | AttributeTargets.Field)]
+    public class ClusteringKeyAttribute : Attribute
+    {
+        public ClusteringKeyAttribute()
+        {
+        }
+    }
 }

shared/JiShe.CollectBus.Common/Attributes/IgnoreJobAttribute.cs

@@ -1,13 +0,0 @@
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
-
-namespace JiShe.CollectBus.Common.Attributes
-{
-    [AttributeUsage(AttributeTargets.Class, Inherited = false)]
-    public class IgnoreJobAttribute : Attribute
-    {
-    }
-}

shared/JiShe.CollectBus.Common/Attributes/NumericalOrderAttribute.cs

@@ -4,7 +4,7 @@ using System.Linq;
 using System.Text;
 using System.Threading.Tasks;
 
-namespace JiShe.CollectBus.Common.AttributeInfo
+namespace JiShe.CollectBus.Common.Attributes
 {
     /// <summary>
     /// 排序序号

shared/JiShe.CollectBus.Common/Attributes/TopicNameAttribute.cs

@@ -1,23 +0,0 @@
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
-using Volo.Abp.EventBus;
-using Volo.Abp;
-
-namespace JiShe.CollectBus.Common.Attributes
-{
-    [AttributeUsage(AttributeTargets.Class, Inherited = false)]
-    public class TopicNameAttribute : Attribute
-    {
-        public virtual string Name { get; }
-
-        public TopicNameAttribute(string name)
-        {
-            this.Name = Check.NotNullOrWhiteSpace(name, nameof(name));
-        }
-
-        public string GetName(Type eventType) => this.Name;
-    }
-}

shared/JiShe.CollectBus.Common/Helpers/CommonHelper.cs

@@ -8,7 +8,7 @@ using System.Runtime.InteropServices;
 using System.Security.Cryptography;
 using System.Text;
 using System.Threading.Tasks;
-using JiShe.CollectBus.Common.AttributeInfo;
+using JiShe.CollectBus.Common.Attributes;
 
 namespace JiShe.CollectBus.Common.Helpers
 {

shared/JiShe.CollectBus.Domain.Shared/CassandraBaseEntity.cs

@@ -0,0 +1,14 @@
+using System;
+using System.ComponentModel.DataAnnotations;
+using JiShe.CollectBus.Common.Attributes;
+
+namespace JiShe.CollectBus
+{
+    public class CassandraBaseEntity<TKey>: ICassandraEntity<TKey>
+    {
+        /// <summary>
+        /// Id
+        /// </summary>
+        public TKey Id { get; set; }
+    }
+}

shared/JiShe.CollectBus.Domain.Shared/ICassandraEntity.cs

@@ -0,0 +1,19 @@
+using System;
+
+namespace JiShe.CollectBus
+{
+    public interface ICassandraEntity<TKey>
+    {
+        TKey Id { get; set; }
+    }
+
+    public interface IHasCreationTime
+    {
+        DateTime CreationTime { get; set; }
+    }
+
+    public interface IHasLastModificationTime
+    {
+        DateTime? LastModificationTime { get; set; }
+    }
+}

shared/JiShe.CollectBus.Domain.Shared/JiShe.CollectBus.Domain.Shared.csproj

@@ -31,4 +31,8 @@
     <None Remove="JiShe.CollectBus.Domain.Shared.abppkg" />
   </ItemGroup>
 
+  <ItemGroup>
+    <ProjectReference Include="..\JiShe.CollectBus.Common\JiShe.CollectBus.Common.csproj" />
+  </ItemGroup>
+
 </Project>

web/JiShe.CollectBus.Host/appsettings.json

@@ -84,7 +84,7 @@
     "SaslPassword": "lixiao1980",
     "KafkaReplicationFactor": 3,
     "NumPartitions": 30,
-    "ServerTagName": "JiSheCollectBus2"
+    "ServerTagName": "JiSheCollectBus20"
   },
   "IoTDBOptions": {
     "UserName": "root",
@@ -95,7 +95,6 @@
     "OpenDebugMode": true,
     "UseTableSessionPoolByDefault": false
   }, 
-  "ServerTagName": "JiSheCollectBus3", 
   "Cassandra": {
     "ReplicationStrategy": {
       "Class": "NetworkTopologyStrategy", //策略为NetworkTopologyStrategy时才会有多个数据中心，SimpleStrategy用在只有一个数据中心的情况下