diff --git a/src/JiShe.CollectBus.Application/CollectBusApplicationModule.cs b/src/JiShe.CollectBus.Application/CollectBusApplicationModule.cs
index 87d63d0..453855d 100644
--- a/src/JiShe.CollectBus.Application/CollectBusApplicationModule.cs
+++ b/src/JiShe.CollectBus.Application/CollectBusApplicationModule.cs
@@ -29,6 +29,8 @@ using Volo.Abp.EventBus.Kafka;
 using Volo.Abp.Kafka;
 using Volo.Abp.EventBus;
 using Confluent.Kafka;
+using JiShe.CollectBus.Kafka.AdminClient;
+using JiShe.CollectBus.Common.Consts;
 
 namespace JiShe.CollectBus;
 
@@ -98,12 +100,24 @@ public class CollectBusApplicationModule : AbpModule
         {
             context.AddBackgroundWorkerAsync(type);
         }
-        
-        var dbContext = context.ServiceProvider.GetRequiredService<EnergySystemScheduledMeterReadingService>();
 
         //默认初始化表计信息
+        var dbContext = context.ServiceProvider.GetRequiredService<EnergySystemScheduledMeterReadingService>();
         dbContext.InitAmmeterCacheData().ConfigureAwait(false).GetAwaiter().GetResult(); 
         dbContext.InitWatermeterCacheData().ConfigureAwait(false).GetAwaiter().GetResult();
+
+        //初始化主题信息
+        var kafkaAdminClient = context.ServiceProvider.GetRequiredService<IAdminClientService>();
+        var configuration = context.ServiceProvider.GetRequiredService<IConfiguration>();
+
+        List<string> topics = ProtocolConstExtensions.GetAllTopicNamesByIssued();
+        topics.AddRange(ProtocolConstExtensions.GetAllTopicNamesByReceived());
+
+        List<TopicSpecification> topicSpecifications = new List<TopicSpecification>();
+        foreach (var item in topics)
+        {
+            kafkaAdminClient.CreateTopicIfNotExistAsync(item, configuration.GetValue<short>(CommonConst.KafkaReplicationFactor), configuration.GetValue<int>(CommonConst.NumPartitions));
+        }
     }
 
 }
diff --git a/src/JiShe.CollectBus.Application/JiShe.CollectBus.Application.csproj b/src/JiShe.CollectBus.Application/JiShe.CollectBus.Application.csproj
index ef887fb..5765ecd 100644
--- a/src/JiShe.CollectBus.Application/JiShe.CollectBus.Application.csproj
+++ b/src/JiShe.CollectBus.Application/JiShe.CollectBus.Application.csproj
@@ -28,6 +28,7 @@
 
 		<ProjectReference Include="..\JiShe.CollectBus.Application.Contracts\JiShe.CollectBus.Application.Contracts.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" />
     <ProjectReference Include="..\JiShe.CollectBus.Protocol.Contracts\JiShe.CollectBus.Protocol.Contracts.csproj" />
   </ItemGroup>
diff --git a/src/JiShe.CollectBus.Common/Consts/CommonConst.cs b/src/JiShe.CollectBus.Common/Consts/CommonConst.cs
index ff38f22..471c897 100644
--- a/src/JiShe.CollectBus.Common/Consts/CommonConst.cs
+++ b/src/JiShe.CollectBus.Common/Consts/CommonConst.cs
@@ -20,5 +20,16 @@ namespace JiShe.CollectBus.Common.Consts
         /// Kafka
         /// </summary>
         public const string Kafka = "Kafka";
+
+        /// <summary>
+        ///  Kafka副本数量
+        /// </summary>
+        public const string KafkaReplicationFactor = "KafkaReplicationFactor";
+
+        /// <summary>
+        ///  Kafka主题分区数量
+        /// </summary>
+        public const string NumPartitions = "NumPartitions";
+
     }
 }
diff --git a/src/JiShe.CollectBus.Host/CollectBusHostModule.cs b/src/JiShe.CollectBus.Host/CollectBusHostModule.cs
index 31f2651..aabc2ba 100644
--- a/src/JiShe.CollectBus.Host/CollectBusHostModule.cs
+++ b/src/JiShe.CollectBus.Host/CollectBusHostModule.cs
@@ -45,7 +45,7 @@ namespace JiShe.CollectBus.Host
             ConfigureHangfire(context);
             ConfigureCap(context, configuration);
             //ConfigureMassTransit(context, configuration);
-            ConfigureKafkaTopic(context, configuration);
+            //ConfigureKafkaTopic(context, configuration);
             ConfigureAuditLog(context);
             ConfigureCustom(context, configuration);
         }
diff --git a/src/JiShe.CollectBus.Host/appsettings.json b/src/JiShe.CollectBus.Host/appsettings.json
index 2c72f7d..4327ff5 100644
--- a/src/JiShe.CollectBus.Host/appsettings.json
+++ b/src/JiShe.CollectBus.Host/appsettings.json
@@ -127,5 +127,7 @@
     "OpenDebugMode": true,
     "UseTableSessionPoolByDefault": false
   },
-  "ServerTagName": "JiSheCollectBus"
+  "ServerTagName": "JiSheCollectBus",
+  "KafkaReplicationFactor": 3,
+  "NumPartitions": 60
 }
\ No newline at end of file