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

添加Header消息过滤开关

Browse Source
This commit is contained in:
zenghongyao 2025-04-16 20:41:52 +08:00
parent 78f9ef349a
commit 3790f3918e
2 changed files with 37 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
src/JiShe.CollectBus.Host/appsettings.json
View File

@ -83,6 +83,7 @@
},
"Kafka": {
"BootstrapServers": "192.168.1.9:29092,192.168.1.9:39092,192.168.1.9:49092",
"EnableFilter": true,
"EnableAuthorization": false,
"SecurityProtocol": "SASL_PLAINTEXT",
"SaslMechanism": "PLAIN",

16
src/JiShe.CollectBus.KafkaProducer/Consumer/ConsumerService.cs
View File

@ -140,16 +140,17 @@ namespace JiShe.CollectBus.Kafka.Consumer
await Task.Delay(TimeSpan.FromSeconds(1),cts.Token);
continue;
}
if (bool.Parse(_configuration["KafkaConsumer:EnableFilter"]!))
{
var headersFilter = new HeadersFilter { { "route-key", Encoding.UTF8.GetBytes(_configuration["ServerTagName"]!) } };
// 检查 Header 是否符合条件
if (!headersFilter.Match(result.Message.Headers))
{
_logger.LogInformation("Kafka消费: {Topic} 分区 {Partition} Header未匹配", result.Topic, result.Partition);
//consumer.Commit(result); // 提交偏移量
// 跳过消息
continue;
}
}
bool sucess= await messageHandler(result.Message.Key, result.Message.Value);
if (sucess)
{
@ -207,15 +208,17 @@ namespace JiShe.CollectBus.Kafka.Consumer
await Task.Delay(100, cts.Token);
continue;
}
if (bool.Parse(_configuration["KafkaConsumer:EnableFilter"]!))
{
var headersFilter = new HeadersFilter { { "route-key", Encoding.UTF8.GetBytes(_configuration["ServerTagName"]!) } };
// 检查 Header 是否符合条件
if (!headersFilter.Match(result.Message.Headers))
{
_logger.LogInformation("Kafka消费: {Topic} 分区 {Partition} Header值未匹配", result.Topic, result.Partition);
//consumer.Commit(result); // 提交偏移量
// 跳过消息
continue;
}
}
bool sucess = await messageHandler(result.Message.Value);
if (sucess)
consumer.Commit(result); // 手动提交
@ -292,16 +295,18 @@ namespace JiShe.CollectBus.Kafka.Consumer
await Task.Delay(TimeSpan.FromSeconds(1), cts.Token);
}
else if (result.Message.Value != null)
{
if (bool.Parse(_configuration["KafkaConsumer:EnableFilter"]!))
{
var headersFilter = new HeadersFilter { { "route-key", Encoding.UTF8.GetBytes(_configuration["ServerTagName"]!) } };
// 检查 Header 是否符合条件
if (!headersFilter.Match(result.Message.Headers))
{
_logger.LogInformation("Kafka消费: {Topic} 分区 {Partition} Header未匹配", result.Topic, result.Partition);
//consumer.Commit(result); // 提交偏移量
// 跳过消息
continue;
}
}
messages.Add((result.Message.Value, result.TopicPartitionOffset));
//messages.Add(result.Message.Value);
}
@ -424,6 +429,8 @@ namespace JiShe.CollectBus.Kafka.Consumer
await Task.Delay(TimeSpan.FromSeconds(1), cts.Token);
}
else if (result.Message.Value != null)
{
if (bool.Parse(_configuration["KafkaConsumer:EnableFilter"]!))
{
var headersFilter = new HeadersFilter { { "route-key", Encoding.UTF8.GetBytes(_configuration["ServerTagName"]!) } };
// 检查 Header 是否符合条件
@ -433,6 +440,7 @@ namespace JiShe.CollectBus.Kafka.Consumer
// 跳过消息
continue;
}
}
messages.Add((result.Message.Value, result.TopicPartitionOffset));
//messages.Add(result.Message.Value);
}