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

@@ -69,8 +69,7 @@ namespace JiShe.CollectBus.Kafka.Producer
             {
                 BootstrapServers = _kafkaOptionConfig.BootstrapServers,
                 //AllowAutoCreateTopics = true,
-                QueueBufferingMaxKbytes = 2097151, // 4194304,  // 修改缓冲区最大为2GB，默认为1GB
-                QueueBufferingMaxMessages = 100000,  // 修改缓冲区消息最大为100000条
+                QueueBufferingMaxKbytes = 4194304,  // 修改缓冲区最大为2GB，默认为1GB
                 CompressionType = CompressionType.Lz4, // 配置使用压缩算法LZ4，其他：gzip/snappy/zstd
                 BatchSize = 32_768, // 修改批次大小为32K
                 LingerMs = 10, // 修改等待时间为20ms，默认为5ms
@@ -142,6 +141,7 @@ namespace JiShe.CollectBus.Kafka.Producer
             var producer = GetProducer<Null, TValue>(typeKey);
             var message = new Message<Null, TValue>
             {
+                //Key= _kafkaOptionConfig.ServerTagName,
                 Value = value,
                 Headers = new Headers{
                    { "route-key", Encoding.UTF8.GetBytes(_applicationOptions.ServerTagName) }
@@ -200,6 +200,7 @@ namespace JiShe.CollectBus.Kafka.Producer
         {
             var message = new Message<Null, TValue>
             {
+                //Key = _kafkaOptionConfig.ServerTagName,
                 Value = value,
                 Headers = new Headers{
                    { "route-key", Encoding.UTF8.GetBytes(_applicationOptions.ServerTagName) }

services/JiShe.CollectBus.Application/DataChannels/DataChannelManageService.cs

@@ -173,7 +173,7 @@ namespace JiShe.CollectBus.DataChannels
                 }
                 catch (Exception ex)
                 {
-                    _logger.LogWarning(ex, "Kafka推送{topicName}主题分区{partition}重试中({Retry}/{MaxRetries})", topicName, partition, retry + 1, maxRetries);
+                    _logger.LogWarning(ex, "Kafka推送{topicName}主题重试中({Retry}/{MaxRetries})", topicName, retry + 1, maxRetries);
                     if (retry == maxRetries - 1) throw;
                     await Task.Delay(1000 * (retry + 1));
                 }

web/JiShe.CollectBus.Host/appsettings.json

@@ -80,6 +80,7 @@
     "SaslPassword": "lixiao1980",
     "KafkaReplicationFactor": 3,
     "NumPartitions": 30,
+    "ServerTagName": "JiSheCollectBus4",
     "FirstCollectionTime": "2025-04-22 16:07:00"
   },
   "IoTDBOptions": {
@@ -141,7 +142,7 @@
     }
   },
   "ServerApplicationOptions": {
-    "ServerTagName": "JiSheCollectBus99",
+    "ServerTagName": "JiSheCollectBus4",
     "SystemType": "Energy",
     "FirstCollectionTime": "2025-04-28 15:07:00",
     "AutomaticVerificationTime": "16:07:00",