合并

JiShe.CollectBus.sln

@@ -37,6 +37,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "JiShe.CollectBus.IoTDBProvi
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "JiShe.CollectBus.Protocol.Test", "src\JiShe.CollectBus.Protocol.Test\JiShe.CollectBus.Protocol.Test.csproj", "{A377955E-7EA1-6F29-8CF7-774569E93925}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "JiShe.CollectBus.Cassandra", "src\JiShe.CollectBus.Cassandra\JiShe.CollectBus.Cassandra.csproj", "{443B4549-0AC0-4493-8F3E-49C83225DD76}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -107,6 +109,10 @@ Global
 		{A377955E-7EA1-6F29-8CF7-774569E93925}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{A377955E-7EA1-6F29-8CF7-774569E93925}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{A377955E-7EA1-6F29-8CF7-774569E93925}.Release|Any CPU.Build.0 = Release|Any CPU
+		{443B4549-0AC0-4493-8F3E-49C83225DD76}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{443B4549-0AC0-4493-8F3E-49C83225DD76}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{443B4549-0AC0-4493-8F3E-49C83225DD76}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{443B4549-0AC0-4493-8F3E-49C83225DD76}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -128,6 +134,7 @@ Global
 		{F0288175-F0EC-48BD-945F-CF1512850943} = {649A3FFA-182F-4E56-9717-E6A9A2BEC545}
 		{A3F3C092-0A25-450B-BF6A-5983163CBEF5} = {649A3FFA-182F-4E56-9717-E6A9A2BEC545}
 		{A377955E-7EA1-6F29-8CF7-774569E93925} = {649A3FFA-182F-4E56-9717-E6A9A2BEC545}
+		{443B4549-0AC0-4493-8F3E-49C83225DD76} = {649A3FFA-182F-4E56-9717-E6A9A2BEC545}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {4324B3B4-B60B-4E3C-91D8-59576B4E26DD}

src/JiShe.CollectBus.Application/CollectBusApplicationModule.cs

@@ -12,8 +12,11 @@ using Microsoft.Extensions.DependencyInjection;
 using System.Collections.Generic;
 using System.Linq;
 using System.Reflection;
+using System.Threading.Tasks;
+using JiShe.CollectBus.Cassandra;
 using Volo.Abp;
 using Volo.Abp.Application;
+using Volo.Abp.Autofac;
 using Volo.Abp.AutoMapper;
 using Volo.Abp.BackgroundWorkers;
 using Volo.Abp.BackgroundWorkers.Hangfire;
@@ -27,11 +30,13 @@ namespace JiShe.CollectBus;
     typeof(CollectBusApplicationContractsModule),
     typeof(AbpDddApplicationModule),
     typeof(AbpAutoMapperModule),
+    typeof(AbpAutofacModule),
     typeof(AbpBackgroundWorkersHangfireModule),
     typeof(CollectBusFreeRedisModule),
     typeof(CollectBusFreeSqlModule),
     typeof(CollectBusKafkaModule),
-    typeof(CollectBusIoTDBModule)
+    typeof(CollectBusIoTDBModule),
+    typeof(CollectBusCassandraModule)
     )]
 public class CollectBusApplicationModule : AbpModule
 {
@@ -46,20 +51,20 @@ public class CollectBusApplicationModule : AbpModule
         }); 
     }
 
-    public override void OnApplicationInitialization(
+    public override async Task OnApplicationInitializationAsync(
         ApplicationInitializationContext context)
     { 
         var assembly = Assembly.GetExecutingAssembly();
         var types = assembly.GetTypes().Where(t => typeof(ICollectWorker).IsAssignableFrom(t) && !t.IsInterface).ToList();
         foreach (var type in types)
         {
-            context.AddBackgroundWorkerAsync(type);
+            await context.AddBackgroundWorkerAsync(type);
         }
 
         //默认初始化表计信息
         var dbContext = context.ServiceProvider.GetRequiredService<EnergySystemScheduledMeterReadingService>();
-        dbContext.InitAmmeterCacheData().ConfigureAwait(false).GetAwaiter().GetResult(); 
+        await dbContext.InitAmmeterCacheData();
-        //dbContext.InitWatermeterCacheData().ConfigureAwait(false).GetAwaiter().GetResult();
+        //await dbContext.InitWatermeterCacheData();
 
         //初始化主题信息
         var kafkaAdminClient = context.ServiceProvider.GetRequiredService<IAdminClientService>();
@@ -70,7 +75,7 @@ public class CollectBusApplicationModule : AbpModule
 
         foreach (var item in topics)
         {
-            kafkaAdminClient.CreateTopicAsync(item, configuration.GetValue<int>(CommonConst.NumPartitions), configuration.GetValue<short>(CommonConst.KafkaReplicationFactor));
+            await kafkaAdminClient.CreateTopicAsync(item, configuration.GetValue<int>(CommonConst.NumPartitions), configuration.GetValue<short>(CommonConst.KafkaReplicationFactor));
         }
     }
 

src/JiShe.CollectBus.Application/JiShe.CollectBus.Application.csproj

@@ -24,8 +24,11 @@
 		<PackageReference Include="TouchSocket" Version="3.0.19" />
     <PackageReference Include="TouchSocket.Hosting" Version="3.0.19" />
     <PackageReference Include="DotNetCore.CAP" Version="8.3.1" />
+    <PackageReference Include="Volo.Abp.EventBus.Kafka" Version="8.3.3" />
 
 		<ProjectReference Include="..\JiShe.CollectBus.Application.Contracts\JiShe.CollectBus.Application.Contracts.csproj" />
+
+		<ProjectReference Include="..\JiShe.CollectBus.Cassandra\JiShe.CollectBus.Cassandra.csproj" />
     <ProjectReference Include="..\JiShe.CollectBus.FreeRedisProvider\JiShe.CollectBus.FreeRedisProvider.csproj" />
     <ProjectReference Include="..\JiShe.CollectBus.KafkaProducer\JiShe.CollectBus.Kafka.csproj" />
     <ProjectReference Include="..\JiShe.CollectBus.MongoDB\JiShe.CollectBus.MongoDB.csproj" />

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

@@ -0,0 +1,51 @@
+using System;
+using System.Collections.Generic;
+using System.Threading.Tasks;
+using Apache.IoTDB.DataStructure;
+using Apache.IoTDB;
+using Confluent.Kafka;
+using JiShe.CollectBus.Ammeters;
+using JiShe.CollectBus.FreeSql;
+using JiShe.CollectBus.IoTDBProvider;
+using JiShe.CollectBus.IotSystems.PrepayModel;
+using Microsoft.AspNetCore.Authorization;
+using Microsoft.AspNetCore.Mvc;
+using Microsoft.Extensions.Options;
+using JiShe.CollectBus.IoTDBProvider.Context;
+using Microsoft.Extensions.Logging;
+using JiShe.CollectBus.Common.Helpers;
+using JiShe.CollectBus.IotSystems.AFNEntity;
+using JiShe.CollectBus.Protocol.Contracts.Interfaces;
+using Microsoft.Extensions.DependencyInjection;
+using JiShe.CollectBus.Cassandra;
+using JiShe.CollectBus.Common.Enums;
+using JiShe.CollectBus.IotSystems.MessageIssueds;
+using Volo.Abp.Application.Services;
+
+namespace JiShe.CollectBus.Samples;
+
+[AllowAnonymous]
+public class TestAppService : CollectBusAppService, IApplicationService
+{
+    private readonly ILogger<TestAppService> _logger;
+    private readonly ICassandraRepository<MessageIssued,string> _messageIssuedCassandraRepository;
+    public TestAppService(
+        ILogger<TestAppService> logger,
+        ICassandraRepository<MessageIssued, string> messageIssuedCassandraRepository
+        )
+    {
+        _logger = logger;
+        _messageIssuedCassandraRepository = messageIssuedCassandraRepository;
+    }
+    public async Task AddMessage()
+    {
+        await _messageIssuedCassandraRepository.InsertAsync(new MessageIssued
+        {
+            ClientId = Guid.NewGuid().ToString(),
+            Message = Array.Empty<byte>(),
+            DeviceNo = "123321312",
+            MessageId = Guid.NewGuid().ToString(),
+            Type = IssuedEventType.Data
+        });
+    }
+}

src/JiShe.CollectBus.Application/Workers/CreateToBeIssueTaskWorker.cs

@@ -27,7 +27,7 @@ namespace JiShe.CollectBus.Workers
         {
             _logger = logger;
             RecurringJobId = nameof(CreateToBeIssueTaskWorker);
-            CronExpression = $"*/{1} * * * *";
+            CronExpression = "* 10 * * * *";
             this._scheduledMeterReadingService = scheduledMeterReadingService;
         }
 

src/JiShe.CollectBus.Application/Workers/SubscriberFifteenMinuteWorker.cs

@@ -26,7 +26,7 @@ namespace JiShe.CollectBus.Workers
         {
             _logger = logger;
             RecurringJobId = nameof(SubscriberFifteenMinuteWorker);
-            CronExpression = $"*/{15} * * * *";
+            CronExpression = "* 15 * * * *";
             this._scheduledMeterReadingService = scheduledMeterReadingService;
         }
 

src/JiShe.CollectBus.Application/Workers/SubscriberFiveMinuteWorker.cs

@@ -26,7 +26,7 @@ namespace JiShe.CollectBus.Workers
         {
             _logger = logger;
             RecurringJobId = nameof(SubscriberFiveMinuteWorker);
-            CronExpression = $"*/{5} * * * *";
+            CronExpression = "* 5 * * * *";
             this._scheduledMeterReadingService = scheduledMeterReadingService;
         }
 

src/JiShe.CollectBus.Application/Workers/SubscriberOneMinuteWorker.cs

@@ -26,7 +26,7 @@ namespace JiShe.CollectBus.Workers
         {
             _logger = logger;
             RecurringJobId = nameof(SubscriberOneMinuteWorker);
-            CronExpression = $"*/{1} * * * *";
+            CronExpression = "* 1 * * * *";
             this._scheduledMeterReadingService = scheduledMeterReadingService;
         }
 

src/JiShe.CollectBus.Cassandra/CassandraConfig.cs

@@ -0,0 +1,64 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace JiShe.CollectBus.Cassandra
+{
+    public class CassandraConfig
+    {
+        public Node[] Nodes { get; set; }
+        public string Username { get; set; }
+        public string Password { get; set; }
+        public string Keyspace { get; set; }
+        public string ConsistencyLevel { get; set; }
+        public Pooling PoolingOptions { get; set; }
+        public Socket SocketOptions { get; set; }
+        public Query QueryOptions { get; set; }
+
+        public ReplicationStrategy ReplicationStrategy { get; set; }
+    }
+
+    public class Pooling
+    {
+        public int CoreConnectionsPerHost { get; set; }
+        public int MaxConnectionsPerHost { get; set; }
+        public int MaxRequestsPerConnection { get; set; }
+    }
+
+    public class Socket
+    {
+        public int ConnectTimeoutMillis { get; set; }
+        public int ReadTimeoutMillis { get; set; }
+    }
+
+    public class Query
+    {
+        public string ConsistencyLevel { get; set; }
+        public string SerialConsistencyLevel { get; set; }
+        public bool DefaultIdempotence { get; set; }
+    }
+
+    public class ReplicationStrategy
+    {
+        public string Class { get; set; }
+        public DataCenter[] DataCenters { get; set; }
+    }
+
+    public class DataCenter
+    {
+        public string Name { get; set; }
+        public int ReplicationFactor { get; set; }
+        public string Strategy { get; set; }
+    }
+
+    public class Node
+    {
+        public string Host { get; set; }
+        public int Port { get; set; }
+        public string DataCenter { get; set; }
+        public string Rack { get; set; }
+    }
+
+}

src/JiShe.CollectBus.Cassandra/CassandraProvider.cs

@@ -0,0 +1,147 @@
+using System;
+using System.Linq;
+using System.Reflection;
+using System.Text;
+using Cassandra;
+using Cassandra.Mapping;
+using Cassandra.Data.Linq;
+using System.ComponentModel.DataAnnotations;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+using Volo.Abp.DependencyInjection;
+using JiShe.CollectBus.Common.Attributes;
+
+namespace JiShe.CollectBus.Cassandra
+{
+    public class CassandraProvider : IDisposable, ICassandraProvider, ISingletonDependency
+    {
+        private readonly ILogger<CassandraProvider> _logger;
+
+        public Cluster Instance { get; set; }
+
+        public ISession Session { get; set; }
+
+        public CassandraConfig CassandraConfig { get; set; }
+        /// <summary>
+        /// 
+        /// </summary>
+        /// <param name="options"></param>
+        /// <param name="logger"></param>
+        public CassandraProvider(
+            IOptions<CassandraConfig> options,
+            ILogger<CassandraProvider> logger)
+        {
+            CassandraConfig = options.Value;
+            _logger = logger;
+        }
+
+        public void InitClusterAndSession()
+        {
+            GetCluster((keyspace) =>
+            {
+                GetSession(keyspace);
+            });
+        }
+
+        public Cluster GetCluster(Action<string?>? callback=null)
+        {
+            var clusterBuilder = Cluster.Builder();
+
+            // 添加多个节点
+            foreach (var node in CassandraConfig.Nodes)
+            {
+                clusterBuilder.AddContactPoint(node.Host)
+                    .WithPort(node.Port);
+            }
+
+            clusterBuilder.WithCredentials(CassandraConfig.Username, CassandraConfig.Password);
+
+            // 优化连接池配置
+            var poolingOptions = new PoolingOptions()
+                .SetCoreConnectionsPerHost(HostDistance.Local, CassandraConfig.PoolingOptions.CoreConnectionsPerHost)
+                .SetMaxConnectionsPerHost(HostDistance.Local, CassandraConfig.PoolingOptions.MaxConnectionsPerHost)
+                .SetMaxRequestsPerConnection(CassandraConfig.PoolingOptions.MaxRequestsPerConnection)
+                .SetHeartBeatInterval(30000); // 30秒心跳
+
+            clusterBuilder.WithPoolingOptions(poolingOptions);
+
+            // 优化Socket配置
+            var socketOptions = new SocketOptions()
+                .SetConnectTimeoutMillis(CassandraConfig.SocketOptions.ConnectTimeoutMillis)
+                .SetReadTimeoutMillis(CassandraConfig.SocketOptions.ReadTimeoutMillis)
+                .SetTcpNoDelay(true) // 启用Nagle算法
+                .SetKeepAlive(true) // 启用TCP保活
+                .SetReceiveBufferSize(32768) // 32KB接收缓冲区
+                .SetSendBufferSize(32768); // 32KB发送缓冲区
+
+            clusterBuilder.WithSocketOptions(socketOptions);
+
+            // 优化查询选项
+            var queryOptions = new QueryOptions()
+                .SetConsistencyLevel((ConsistencyLevel)Enum.Parse(typeof(ConsistencyLevel), CassandraConfig.QueryOptions.ConsistencyLevel))
+                .SetSerialConsistencyLevel((ConsistencyLevel)Enum.Parse(typeof(ConsistencyLevel), CassandraConfig.QueryOptions.SerialConsistencyLevel))
+                .SetDefaultIdempotence(CassandraConfig.QueryOptions.DefaultIdempotence)
+                .SetPageSize(5000); // 增加页面大小
+
+            clusterBuilder.WithQueryOptions(queryOptions);
+
+            // 启用压缩
+            clusterBuilder.WithCompression(CompressionType.LZ4);
+
+            // 配置重连策略
+            clusterBuilder.WithReconnectionPolicy(new ExponentialReconnectionPolicy(1000, 10 * 60 * 1000));
+            Instance = clusterBuilder.Build();
+            callback?.Invoke(null);
+            return Instance;
+        }
+
+        public ISession GetSession(string? keyspace = null)
+        {
+            if (string.IsNullOrEmpty(keyspace))
+            {
+                keyspace = CassandraConfig.Keyspace;
+            }
+            Session = Instance.Connect();
+            var replication = GetReplicationStrategy();
+            Session.CreateKeyspaceIfNotExists(keyspace, replication);
+            Session.ChangeKeyspace(keyspace);
+            return Session;
+        }
+
+        private Dictionary<string, string> GetReplicationStrategy()
+        {
+            var strategy = CassandraConfig.ReplicationStrategy.Class;
+            var dataCenters = CassandraConfig.ReplicationStrategy.DataCenters;
+
+            switch (strategy)
+            {
+                case "NetworkTopologyStrategy":
+                    var networkDic = new Dictionary<string, string> { { "class", "NetworkTopologyStrategy" } };
+                    foreach (var dataCenter in dataCenters)
+                    {
+                        networkDic.Add(dataCenter.Name, dataCenter.ReplicationFactor.ToString());
+                    }
+                    return networkDic;
+                case "SimpleStrategy":
+                    var dic = new Dictionary<string, string> { { "class", "SimpleStrategy" } };
+                    if (dataCenters.Length >= 1)
+                    {
+                        dic.Add("replication_factor", dataCenters[0].ReplicationFactor.ToString());
+                    }
+                    else
+                    {
+                        _logger.LogError("SimpleStrategy 不支持多个数据中心！");
+                    }
+                    return dic;
+                default:
+                    throw new ArgumentNullException($"Strategy", "Strategy配置错误！");
+            }
+        }
+
+        public void Dispose()
+        {
+            Instance.Dispose();
+            Session.Dispose();
+        }
+    }
+}

src/JiShe.CollectBus.Cassandra/CassandraQueryOptimizer.cs

@@ -0,0 +1,156 @@
+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);
+        }
+    }
+} 

src/JiShe.CollectBus.Cassandra/CassandraRepository.cs

@@ -0,0 +1,69 @@
+using Cassandra;
+using Cassandra.Mapping;
+using JiShe.CollectBus.Cassandra.Extensions;
+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
+    {
+
+        public CassandraRepository(ICassandraProvider cassandraProvider, MappingConfiguration mappingConfig)
+        {
+            Mapper = new Mapper(cassandraProvider.Session, mappingConfig);
+            cassandraProvider.Session.CreateTable<TEntity>(cassandraProvider.CassandraConfig.Keyspace);
+        }
+
+        public readonly IMapper Mapper;
+
+        public virtual async Task<TEntity> GetAsync(TKey id)
+        {
+            return await Mapper.SingleOrDefaultAsync<TEntity>("WHERE id = ?", id);
+        }
+
+        public virtual async Task<List<TEntity>> GetListAsync()
+        {
+            return (await Mapper.FetchAsync<TEntity>()).ToList();
+        }
+
+        public virtual async Task<TEntity> InsertAsync(TEntity entity)
+        {
+            await Mapper.InsertAsync(entity);
+            return entity;
+        }
+
+        public virtual async Task<TEntity> UpdateAsync(TEntity entity)
+        {
+            await Mapper.UpdateAsync(entity);
+            return entity;
+        }
+
+        public virtual async Task DeleteAsync(TEntity entity)
+        {
+            await Mapper.DeleteAsync(entity);
+        }
+
+        public virtual async Task DeleteAsync(TKey id)
+        {
+            await Mapper.DeleteAsync<TEntity>("WHERE id = ?", id);
+        }
+
+        public virtual async Task<List<TEntity>> GetPagedListAsync(
+            int skipCount,
+            int maxResultCount,
+            string sorting)
+        {
+            var cql = $"SELECT * FROM {typeof(TEntity).Name.ToLower()}";
+            if (!string.IsNullOrWhiteSpace(sorting))
+            {
+                cql += $" ORDER BY {sorting}";
+            }
+            cql += $" LIMIT {maxResultCount} OFFSET {skipCount}";
+
+            return (await Mapper.FetchAsync<TEntity>(cql)).ToList();
+        }
+    }
+} 

src/JiShe.CollectBus.Cassandra/CollectBusCassandraModule.cs

@@ -0,0 +1,29 @@
+using Cassandra;
+using Cassandra.Mapping;
+using JiShe.CollectBus.Cassandra.Mappers;
+using Microsoft.Extensions.DependencyInjection;
+using Volo.Abp;
+using Volo.Abp.Autofac;
+using Volo.Abp.Modularity;
+
+namespace JiShe.CollectBus.Cassandra
+{
+    [DependsOn(
+        typeof(AbpAutofacModule)
+    )]
+    public class CollectBusCassandraModule : AbpModule
+    {
+        public override void ConfigureServices(ServiceConfigurationContext context)
+        {
+            Configure<CassandraConfig>(context.Services.GetConfiguration().GetSection("Cassandra"));
+
+            context.AddCassandra();
+
+        }
+
+        public override void OnApplicationInitialization(ApplicationInitializationContext context)
+        {
+            context.UseCassandra();
+        }
+    }
+}

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

@@ -0,0 +1,35 @@
+using Autofac.Core;
+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;
+
+// ReSharper disable once CheckNamespace
+namespace Microsoft.Extensions.DependencyInjection
+{
+    public static class ApplicationInitializationContextExtensions
+    {
+        public static void UseCassandra(this ApplicationInitializationContext context)
+        {
+            var service = context.ServiceProvider;
+            var cassandraProvider = service.GetRequiredService<ICassandraProvider>();
+            cassandraProvider.InitClusterAndSession();
+        }
+    }
+
+    public static class ServiceCollectionExtensions
+    {
+        public static void AddCassandra(this ServiceConfigurationContext context)
+        {
+            context.Services.AddSingleton(typeof(ICassandraRepository<,>), typeof(CassandraRepository<,>));
+
+            var mappingConfig = new MappingConfiguration()
+                .Define(new CollectBusMapping());
+            context.Services.AddSingleton(mappingConfig);
+        }
+    }
+}

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

@@ -0,0 +1,83 @@
+using System.Reflection;
+using System.Text;
+using Cassandra;
+using System.ComponentModel.DataAnnotations;
+using JiShe.CollectBus.Common.Attributes;
+using Cassandra.Mapping;
+
+namespace JiShe.CollectBus.Cassandra.Extensions
+{
+    public static class SessionExtension
+    {
+        public static void CreateTable<TEntity>(this ISession session,string? defaultKeyspace=null) where TEntity : class
+        {
+            var type = typeof(TEntity);
+            var tableAttribute = type.GetCustomAttribute<CassandraTableAttribute>();
+            var tableName = tableAttribute?.Name ?? type.Name.ToLower();
+            var tableKeyspace = tableAttribute?.Keyspace ?? defaultKeyspace;
+
+            var properties = type.GetProperties();
+            var primaryKey = properties.FirstOrDefault(p => p.GetCustomAttribute<KeyAttribute>() != null);
+
+            if (primaryKey == null)
+            {
+                throw new InvalidOperationException($"No primary key defined for type {type.Name}");
+            }
+
+            var cql = new StringBuilder();
+            cql.Append($"CREATE TABLE IF NOT EXISTS {tableKeyspace}.{tableName} (");
+
+            foreach (var prop in properties)
+            {
+                var ignoreAttribute = prop.GetCustomAttribute<CassandraIgnoreAttribute>();
+                if (ignoreAttribute != null) continue;
+                var columnName = prop.Name.ToLower();
+                var cqlType = GetCassandraType(prop.PropertyType);
+
+                cql.Append($"{columnName} {cqlType}, ");
+            }
+            cql.Length -= 2; // Remove last comma and space
+            cql.Append($", PRIMARY KEY ({primaryKey.Name.ToLower()}))");
+
+            session.Execute(cql.ToString());
+        }
+
+        private static string GetCassandraType(Type type)
+        {
+            // 基础类型处理
+            switch (Type.GetTypeCode(type))
+            {
+                case TypeCode.String: return "text";
+                case TypeCode.Int32: return "int";
+                case TypeCode.Int64: return "bigint";
+                case TypeCode.Boolean: return "boolean";
+                case TypeCode.DateTime: return "timestamp";
+                case TypeCode.Byte: return "tinyint";
+            }
+
+            if (type == typeof(Guid)) return "uuid";
+            if (type == typeof(DateTimeOffset)) return "timestamp";
+            if (type == typeof(Byte[])) return "blob";
+
+            // 处理集合类型
+            if (type.IsGenericType)
+            {
+                var genericType = type.GetGenericTypeDefinition();
+                var elementType = type.GetGenericArguments()[0];
+
+                if (genericType == typeof(List<>))
+                    return $"list<{GetCassandraType(elementType)}>";
+                if (genericType == typeof(HashSet<>))
+                    return $"set<{GetCassandraType(elementType)}>";
+                if (genericType == typeof(Dictionary<,>))
+                {
+                    var keyType = type.GetGenericArguments()[0];
+                    var valueType = type.GetGenericArguments()[1];
+                    return $"map<{GetCassandraType(keyType)}, {GetCassandraType(valueType)}>";
+                }
+            }
+
+            throw new NotSupportedException($"不支持的类型: {type.Name}");
+        }
+    }
+}

src/JiShe.CollectBus.Cassandra/ICassandraProvider.cs

@@ -0,0 +1,24 @@
+using Cassandra;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace JiShe.CollectBus.Cassandra
+{
+    public interface ICassandraProvider
+    {
+        Cluster Instance { get;}
+
+        ISession Session { get;}
+
+        CassandraConfig CassandraConfig { get; }
+
+        ISession GetSession(string? keyspace = null);
+
+        Cluster GetCluster(Action<string?>? callback = null);
+
+        void InitClusterAndSession();
+    }
+}

src/JiShe.CollectBus.Cassandra/ICassandraRepository.cs

@@ -0,0 +1,20 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using Volo.Abp.Domain.Entities;
+
+namespace JiShe.CollectBus.Cassandra
+{
+    public interface ICassandraRepository<TEntity, TKey> where TEntity : class
+    {
+        Task<TEntity> GetAsync(TKey id);
+        Task<List<TEntity>> GetListAsync();
+        Task<TEntity> InsertAsync(TEntity entity);
+        Task<TEntity> UpdateAsync(TEntity entity);
+        Task DeleteAsync(TEntity entity);
+        Task DeleteAsync(TKey id);
+        Task<List<TEntity>> GetPagedListAsync(int skipCount, int maxResultCount, string sorting);
+    }
+}

src/JiShe.CollectBus.Cassandra/JiShe.CollectBus.Cassandra.csproj

@@ -0,0 +1,22 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <TargetFramework>net8.0</TargetFramework>
+    <ImplicitUsings>enable</ImplicitUsings>
+    <Nullable>enable</Nullable>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="CassandraCSharpDriver" Version="3.22.0" />
+    <PackageReference Include="Microsoft.Extensions.Options" Version="8.0.2" />
+    <PackageReference Include="Volo.Abp.Autofac" Version="8.3.3" />
+    <PackageReference Include="Volo.Abp.Core" Version="8.3.3" />
+    <PackageReference Include="Volo.Abp.Ddd.Domain" Version="8.3.3" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\JiShe.CollectBus.Common\JiShe.CollectBus.Common.csproj" />
+    <ProjectReference Include="..\JiShe.CollectBus.Domain\JiShe.CollectBus.Domain.csproj" />
+  </ItemGroup>
+
+</Project>

src/JiShe.CollectBus.Cassandra/Mappers/CollectBusMapping.cs

@@ -0,0 +1,20 @@
+using Cassandra.Mapping;
+using System;
+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
+{
+    public class CollectBusMapping: Mappings
+    {
+        public CollectBusMapping()
+        {
+            For<MessageIssued>()
+                .Column(e => e.Type, cm => cm.WithName("type").WithDbType<int>());
+        }
+    }
+}

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

@@ -0,0 +1,32 @@
+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 CassandraTableAttribute : Attribute
+    {
+        public CassandraTableAttribute(string? name = null,string? keyspace =null)
+        {
+            Name = name;
+            Keyspace = keyspace;
+        }
+
+        public virtual string? Name { get; }
+
+        public virtual string? Keyspace { get; }
+    }
+
+    [AttributeUsage(AttributeTargets.Property | AttributeTargets.Field)]
+    public class CassandraIgnoreAttribute : Attribute
+    {
+        public CassandraIgnoreAttribute()
+        {
+        }
+    }
+}

src/JiShe.CollectBus.Domain/IotSystems/MessageIssueds/MessageIssued.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.Common.Enums;
 
 namespace JiShe.CollectBus.IotSystems.MessageIssueds
 {
+    [CassandraTable]
     public class MessageIssued
     {
+        [Key]
         public string ClientId { get; set; }
         public byte[] Message { get; set; }
         public string DeviceNo { get; set; }

src/JiShe.CollectBus.Host/Program.cs

@@ -1,12 +1,35 @@
 using JiShe.CollectBus.Host;
 using Microsoft.AspNetCore.Hosting;
 using Serilog;
+using Volo.Abp.Modularity.PlugIns;
 
 public class Program
 {
-    public static void Main(string[] args)
+    /// <summary>
+    /// 
+    /// </summary>
+    /// <param name="args"></param>
+    /// <returns></returns>
+    public static async Task Main(string[] args)
     {
-        CreateHostBuilder(args).Build().Run();
+        var builder = WebApplication.CreateBuilder(args);
+        builder.Host.UseContentRoot(Directory.GetCurrentDirectory())
+            .UseSerilog((context, loggerConfiguration) =>
+            {
+                loggerConfiguration.ReadFrom.Configuration(context.Configuration);
+            })
+            .UseAutofac();
+        await builder.AddApplicationAsync<CollectBusHostModule>(options =>
+        {
+            // 加载插件，固定模式，可热插拔
+            options.PlugInSources.AddFolder(Path.Combine(AppDomain.CurrentDomain.BaseDirectory, "Plugins"));
+        });
+        var app = builder.Build();
+        await app.InitializeApplicationAsync();
+        await app.RunAsync();
+
+
+        //await CreateHostBuilder(args).Build().RunAsync();
     }
 
     private static IHostBuilder CreateHostBuilder(string[] args) =>
@@ -16,11 +39,12 @@ public class Program
             {
                 loggerConfiguration.ReadFrom.Configuration(context.Configuration);
             })
-            .UseAutofac()
             .ConfigureWebHostDefaults(webBuilder =>
             {
                 webBuilder.UseStartup<Startup>();
-            });
+            })
+            .UseAutofac();
+
 
 
 
@@ -34,8 +58,8 @@ public class Program
             });
 
 
-    private static void ConfigureServices(IServiceCollection services, HostBuilderContext hostContext)
+    private static async Task ConfigureServices(IServiceCollection services, HostBuilderContext hostContext)
     {
-        services.AddApplication<CollectBusHostModule>();
+        await services.AddApplicationAsync<CollectBusHostModule>();
     }
 }

src/JiShe.CollectBus.Host/Startup.cs

@@ -39,10 +39,15 @@ namespace JiShe.CollectBus.Host
         /// <param name="lifetime">The lifetime.</param>
         public void Configure(IApplicationBuilder app, IHostApplicationLifetime lifetime)
         {
-            app.InitializeApplication();
-            //await app.InitializeApplicationAsync();
 
+            app.Use(async (context, next) =>
+            {
+                // 在请求处理之前调用 InitializeApplicationAsync
+                await app.InitializeApplicationAsync();
 
+                // 继续请求管道中的下一个中间件
+                await next();
+            });
         }
     }
 }

src/JiShe.CollectBus.Host/appsettings.json

@@ -130,5 +130,48 @@
   },
   "ServerTagName": "JiSheCollectBus",
   "KafkaReplicationFactor": 3,
-  "NumPartitions": 30
+  "NumPartitions": 30,
+  "Cassandra": {
+    "ReplicationStrategy": {
+      "Class": "NetworkTopologyStrategy", //策略为NetworkTopologyStrategy时才会有多个数据中心，SimpleStrategy用在只有一个数据中心的情况下
+      "DataCenters": [
+        {
+          "Name": "dc1",
+          "ReplicationFactor": 3
+        }
+      ]
+    },
+    "Nodes": [
+      {
+        "Host": "192.168.1.9",
+        "Port": 9042,
+        "DataCenter": "dc1",
+        "Rack": "RAC1"
+      },
+      {
+        "Host": "192.168.1.9",
+        "Port": 9043,
+        "DataCenter": "dc1",
+        "Rack": "RAC2"
+      }
+    ],
+    "Username": "admin",
+    "Password": "lixiao1980",
+    "Keyspace": "jishecollectbus",
+    "ConsistencyLevel": "Quorum",
+    "PoolingOptions": {
+      "CoreConnectionsPerHost": 4,
+      "MaxConnectionsPerHost": 8,
+      "MaxRequestsPerConnection": 2000
+    },
+    "SocketOptions": {
+      "ConnectTimeoutMillis": 10000,
+      "ReadTimeoutMillis": 20000
+    },
+    "QueryOptions": {
+      "ConsistencyLevel": "Quorum",
+      "SerialConsistencyLevel": "Serial",
+      "DefaultIdempotence": true
+    }
+  }
 }

src/JiShe.CollectBus.Protocol/JiSheCollectBusProtocolModule.cs

@@ -12,10 +12,10 @@ namespace JiShe.CollectBus.Protocol
             context.Services.AddKeyedSingleton<IProtocolPlugin, StandardProtocolPlugin>(nameof(StandardProtocolPlugin));
         }
 
-        public override void OnApplicationInitialization(ApplicationInitializationContext context)
+        public override async Task OnApplicationInitializationAsync(ApplicationInitializationContext context)
         {
             var standardProtocol = context.ServiceProvider.GetRequiredKeyedService<IProtocolPlugin>(nameof(StandardProtocolPlugin));
-            standardProtocol.AddAsync();
+            await standardProtocol.AddAsync();
         }
     }
 }