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

优化kafka发布消息

Browse Source
This commit is contained in:
zenghongyao 2025-05-06 18:41:23 +08:00
parent c4335aa3b7
commit 1e85f894a8
3 changed files with 28 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -7,6 +7,7 @@ using System.Text;
using System.Threading.Tasks; using System.Threading.Tasks;
using Confluent.Kafka; using Confluent.Kafka;
using JiShe.CollectBus.Common; using JiShe.CollectBus.Common;
using JiShe.CollectBus.Common.Helpers;
using JiShe.CollectBus.Kafka.Consumer; using JiShe.CollectBus.Kafka.Consumer;
using JiShe.CollectBus.Kafka.Internal; using JiShe.CollectBus.Kafka.Internal;
using JiShe.CollectBus.Kafka.Serialization; using JiShe.CollectBus.Kafka.Serialization;
@ -69,11 +70,11 @@ namespace JiShe.CollectBus.Kafka.Producer
{ {
BootstrapServers = _kafkaOptionConfig.BootstrapServers, BootstrapServers = _kafkaOptionConfig.BootstrapServers,
//AllowAutoCreateTopics = true, //AllowAutoCreateTopics = true,
QueueBufferingMaxKbytes = 2097151, // 4194304, // 修改缓冲区最大为2GB，默认为1GB QueueBufferingMaxKbytes = 4_194_304, // 4_194_304 2_097_151 // 修改缓冲区最大为2GB，默认为1GB
QueueBufferingMaxMessages = 100000, // 修改缓冲区消息最大为100000条 QueueBufferingMaxMessages = int.MaxValue, // 修改缓冲区消息条
CompressionType = CompressionType.Lz4, // 配置使用压缩算法LZ4，其他：gzip/snappy/zstd CompressionType = CompressionType.Lz4, // 配置使用压缩算法LZ4，其他：gzip/snappy/zstd
BatchSize = 32_768, // 修改批次大小为32K BatchSize = 32_768, // 修改批次大小为32K
LingerMs = 10, // 修改等待时间为20ms，默认为5ms LingerMs = 20, // 修改等待时间为20ms，默认为5ms
Acks = Acks.All, // 表明只有所有副本Broker都收到消息才算提交成功, 可以 Acks.Leader Acks = Acks.All, // 表明只有所有副本Broker都收到消息才算提交成功, 可以 Acks.Leader
MessageSendMaxRetries = 50, // 消息发送失败最大重试50次 MessageSendMaxRetries = 50, // 消息发送失败最大重试50次
MessageTimeoutMs = 120000, // 消息发送超时时间为2分钟,设置值MessageTimeoutMs > LingerMs MessageTimeoutMs = 120000, // 消息发送超时时间为2分钟,设置值MessageTimeoutMs > LingerMs
@ -175,7 +176,7 @@ namespace JiShe.CollectBus.Kafka.Producer
var producer = GetProducer<TKey, TValue>(typeKey); var producer = GetProducer<TKey, TValue>(typeKey);
if (partition.HasValue) if (partition.HasValue)
{ {
var topicPartition = new TopicPartition(topic, partition.Value); var topicPartition = new TopicPartition(topic, new Partition(partition.Value));
producer.Produce(topicPartition, message, deliveryHandler); producer.Produce(topicPartition, message, deliveryHandler);
} }
else else
@ -209,7 +210,8 @@ namespace JiShe.CollectBus.Kafka.Producer
var producer = GetProducer<Null, TValue>(typeKey); var producer = GetProducer<Null, TValue>(typeKey);
if (partition.HasValue) if (partition.HasValue)
{ {
var topicPartition = new TopicPartition(topic, partition.Value); var topicPartition = new TopicPartition(topic,new Partition(partition.Value));
//_logger.LogError($"push消息：{topic}-{partition.Value}");
producer.Produce(topicPartition, message, deliveryHandler); producer.Produce(topicPartition, message, deliveryHandler);
} }
else else

16
protocols/JiShe.CollectBus.Protocol.T37612012/T37612012ProtocolPlugin.cs
View File

@ -610,7 +610,7 @@ namespace JiShe.CollectBus.Protocol.T37612012
{ {
unitData = new UnitData unitData = new UnitData
{ {
HexMessageList = hexStringList.GetRange(14, hexStringList.Count - 14 - 2) //总数字节数-固定长度报文头-控制域C-地址域A-校验和CS-结束字符（16H） HexMessageList = hexStringList. (14, hexStringList.Count - 14 - 2) //总数字节数-固定长度报文头-控制域C-地址域A-校验和CS-结束字符（16H）
}; };
unitData.HexMessageString = string.Join(" ", unitData.HexMessageList); unitData.HexMessageString = string.Join(" ", unitData.HexMessageList);
} }
@ -715,10 +715,8 @@ namespace JiShe.CollectBus.Protocol.T37612012
/// <returns></returns> /// <returns></returns>
public virtual List<string> Generate_DataUnit(DataTimeMark timeMark) public virtual List<string> Generate_DataUnit(DataTimeMark timeMark)
{ {
List<string> values = new List<string> List<string> values = new List<string>();
{ values.AddRange(SplitDataTime(timeMark.DataTime));//数据时间
SplitDataTime(timeMark.DataTime)//数据时间
};
if (timeMark.Density > 0) if (timeMark.Density > 0)
values.Add(timeMark.Density.HexToDecStr().PadLeft(2, '0'));//密度 values.Add(timeMark.Density.HexToDecStr().PadLeft(2, '0'));//密度
if (timeMark.Point > 0) if (timeMark.Point > 0)
@ -727,13 +725,13 @@ namespace JiShe.CollectBus.Protocol.T37612012
} }
private string SplitDataTime(DateTime dataTime) private List<string> SplitDataTime(DateTime dataTime)
{ {
//2101060815 //2101060815
List<string> values = new List<string>() { $"{dataTime}:YY", $"{dataTime}:MM", $"{dataTime}:dd", $"{dataTime}:HH", $"{dataTime}:mm", }; List<string> values = new List<string>() { $"{dataTime:yy}", $"{dataTime:MM}", $"{dataTime:dd}", $"{dataTime:HH}", $"{dataTime:mm}", };
values.Reverse(); values.Reverse();
return string.Join("", values); return values;
//return string.Join("", values);
} }
#endregion #endregion

8
shared/JiShe.CollectBus.Common/BuildSendDatas/Build3761SendData.cs
View File

@ -1605,6 +1605,8 @@ namespace JiShe.CollectBus.Common.BuildSendDatas
/// <param name="userData">用户数据区</param> /// <param name="userData">用户数据区</param>
/// <returns></returns> /// <returns></returns>
private static string GetCS(List<string> userData) private static string GetCS(List<string> userData)
{
try
{ {
byte sum = 0; byte sum = 0;
foreach (var d in userData) foreach (var d in userData)
@ -1614,6 +1616,12 @@ namespace JiShe.CollectBus.Common.BuildSendDatas
} }
return sum.ToString("X2"); return sum.ToString("X2");
} }
catch(Exception)
{
throw;
}
}
/// <summary> /// <summary>
/// 用户数据区 /// 用户数据区