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-04-19 00:30:58 +08:00
parent dc7416bdbf
commit 270abd656f
26 changed files with 519 additions and 225 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
JiShe.CollectBus.sln
View File

@ -47,6 +47,8 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "2.Services", "2.Services",
EndProject
Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "5.Shared", "5.Shared", "{EBF7C01F-9B4F-48E6-8418-2CBFDA51EB0B}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "JiShe.CollectBus.Kafka.Test", "modules\JiShe.CollectBus.Kafka.Test\JiShe.CollectBus.Kafka.Test.csproj", "{6D6A2A58-7406-9C8C-7B23-3E442CCE3E6B}"
EndProject
Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution
Debug|Any CPU = Debug|Any CPU
@ -121,6 +123,10 @@ Global
{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
{6D6A2A58-7406-9C8C-7B23-3E442CCE3E6B}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{6D6A2A58-7406-9C8C-7B23-3E442CCE3E6B}.Debug|Any CPU.Build.0 = Debug|Any CPU
{6D6A2A58-7406-9C8C-7B23-3E442CCE3E6B}.Release|Any CPU.ActiveCfg = Release|Any CPU
{6D6A2A58-7406-9C8C-7B23-3E442CCE3E6B}.Release|Any CPU.Build.0 = Release|Any CPU
EndGlobalSection
GlobalSection(SolutionProperties) = preSolution
HideSolutionNode = FALSE
@ -143,6 +149,7 @@ Global
{A3F3C092-0A25-450B-BF6A-5983163CBEF5} = {2E0FE301-34C3-4561-9CAE-C7A9E65AEE59}
{A377955E-7EA1-6F29-8CF7-774569E93925} = {3C3F9DB2-EC97-4464-B49F-BF1A0C2B46DC}
{443B4549-0AC0-4493-8F3E-49C83225DD76} = {2E0FE301-34C3-4561-9CAE-C7A9E65AEE59}
{6D6A2A58-7406-9C8C-7B23-3E442CCE3E6B} = {2E0FE301-34C3-4561-9CAE-C7A9E65AEE59}
EndGlobalSection
GlobalSection(ExtensibilityGlobals) = postSolution
SolutionGuid = {4324B3B4-B60B-4E3C-91D8-59576B4E26DD}

76
modules/JiShe.CollectBus.Kafka.Test/KafkaSubscribeTest.cs
View File

@ -3,6 +3,7 @@ using JiShe.CollectBus.Common.Enums;
using JiShe.CollectBus.Common.Models;
using JiShe.CollectBus.IotSystems.MessageReceiveds;
using JiShe.CollectBus.Kafka.Attributes;
using JiShe.CollectBus.Kafka.Internal;
using Microsoft.Extensions.DependencyInjection;
using System;
using System.Collections.Generic;
@ -16,53 +17,54 @@ namespace JiShe.CollectBus.Kafka.Test
{
public class KafkaSubscribeTest: IKafkaSubscribe
{
[KafkaSubscribe(ProtocolConst.TESTTOPIC, EnableBatch=false,BatchSize=1000)]
[KafkaSubscribe(ProtocolConst.TESTTOPIC, EnableBatch = false, BatchSize = 10)]
public async Task<ISubscribeAck> KafkaSubscribeAsync(object obj)
public async Task<ISubscribeAck> KafkaSubscribeAsync(TestTopic obj)
//public async Task<ISubscribeAck> KafkaSubscribeAsync(IEnumerable<int> obj)
{
Console.WriteLine($"收到订阅消息： {JsonSerializer.Serialize(obj)}");
return SubscribeAck.Success();
}
[KafkaSubscribe(ProtocolConst.SubscriberLoginIssuedEventName)]
//[CapSubscribe(ProtocolConst.SubscriberLoginIssuedEventName)]
public async Task<ISubscribeAck> LoginIssuedEvent(IssuedEventMessage issuedEventMessage)
{
Console.WriteLine($"收到订阅消息： {JsonSerializer.Serialize(issuedEventMessage)}");
return SubscribeAck.Success();
}
//[KafkaSubscribe(ProtocolConst.SubscriberLoginIssuedEventName)]
////[CapSubscribe(ProtocolConst.SubscriberLoginIssuedEventName)]
//public async Task<ISubscribeAck> LoginIssuedEvent(IssuedEventMessage issuedEventMessage)
//{
// Console.WriteLine($"收到订阅消息： {JsonSerializer.Serialize(issuedEventMessage)}");
// return SubscribeAck.Success();
//}
[KafkaSubscribe(ProtocolConst.SubscriberHeartbeatIssuedEventName)]
//[CapSubscribe(ProtocolConst.SubscriberHeartbeatIssuedEventName)]
public async Task<ISubscribeAck> HeartbeatIssuedEvent(IssuedEventMessage issuedEventMessage)
{
Console.WriteLine($"收到订阅消息： {JsonSerializer.Serialize(issuedEventMessage)}");
return SubscribeAck.Success();
}
//[KafkaSubscribe(ProtocolConst.SubscriberHeartbeatIssuedEventName)]
////[CapSubscribe(ProtocolConst.SubscriberHeartbeatIssuedEventName)]
//public async Task<ISubscribeAck> HeartbeatIssuedEvent(IssuedEventMessage issuedEventMessage)
//{
// Console.WriteLine($"收到订阅消息： {JsonSerializer.Serialize(issuedEventMessage)}");
// return SubscribeAck.Success();
//}
[KafkaSubscribe(ProtocolConst.SubscriberReceivedEventName)]
//[CapSubscribe(ProtocolConst.SubscriberReceivedEventName)]
public async Task<ISubscribeAck> ReceivedEvent(MessageReceived receivedMessage)
{
Console.WriteLine($"收到订阅消息： {JsonSerializer.Serialize(receivedMessage)}");
return SubscribeAck.Success();
}
//[KafkaSubscribe(ProtocolConst.SubscriberReceivedEventName)]
////[CapSubscribe(ProtocolConst.SubscriberReceivedEventName)]
//public async Task<ISubscribeAck> ReceivedEvent(MessageReceived receivedMessage)
//{
// Console.WriteLine($"收到订阅消息： {JsonSerializer.Serialize(receivedMessage)}");
// return SubscribeAck.Success();
//}
[KafkaSubscribe(ProtocolConst.SubscriberHeartbeatReceivedEventName)]
//[CapSubscribe(ProtocolConst.SubscriberHeartbeatReceivedEventName)]
public async Task<ISubscribeAck> ReceivedHeartbeatEvent(MessageReceivedHeartbeat receivedHeartbeatMessage)
{
Console.WriteLine($"收到订阅消息： {JsonSerializer.Serialize(receivedHeartbeatMessage)}");
return SubscribeAck.Success();
}
//[KafkaSubscribe(ProtocolConst.SubscriberHeartbeatReceivedEventName)]
////[CapSubscribe(ProtocolConst.SubscriberHeartbeatReceivedEventName)]
//public async Task<ISubscribeAck> ReceivedHeartbeatEvent(MessageReceivedHeartbeat receivedHeartbeatMessage)
//{
// Console.WriteLine($"收到订阅消息： {JsonSerializer.Serialize(receivedHeartbeatMessage)}");
// return SubscribeAck.Success();
//}
[KafkaSubscribe(ProtocolConst.SubscriberLoginReceivedEventName)]
//[CapSubscribe(ProtocolConst.SubscriberLoginReceivedEventName)]
public async Task<ISubscribeAck> ReceivedLoginEvent(MessageReceivedLogin receivedLoginMessage)
{
Console.WriteLine($"收到订阅消息： {JsonSerializer.Serialize(receivedLoginMessage)}");
return SubscribeAck.Success();
}
//[KafkaSubscribe(ProtocolConst.SubscriberLoginReceivedEventName)]
////[CapSubscribe(ProtocolConst.SubscriberLoginReceivedEventName)]
//public async Task<ISubscribeAck> ReceivedLoginEvent(MessageReceivedLogin receivedLoginMessage)
//{
// Console.WriteLine($"收到订阅消息： {JsonSerializer.Serialize(receivedLoginMessage)}");
// return SubscribeAck.Success();
//}
}
}

34
modules/JiShe.CollectBus.Kafka.Test/Program.cs
View File

@ -7,6 +7,7 @@ using JiShe.CollectBus.Common.Consts;
using JiShe.CollectBus.Kafka;
using JiShe.CollectBus.Kafka.AdminClient;
using JiShe.CollectBus.Kafka.Consumer;
using JiShe.CollectBus.Kafka.Internal;
using JiShe.CollectBus.Kafka.Producer;
using JiShe.CollectBus.Kafka.Test;
using Microsoft.AspNetCore.Builder;
@ -86,12 +87,13 @@ var logger = loggerFactory.CreateLogger<Program>();
logger.LogInformation("程序启动");
var adminClientService = host.Services.GetRequiredService<IAdminClientService>();
var configuration = host.Services.GetRequiredService<IConfiguration>();
string topic = "test-topic";
string topic = ProtocolConst.TESTTOPIC;
//await adminClientService.DeleteTopicAsync(topic);
// 创建 topic
//await adminClientService.CreateTopicAsync(topic, configuration.GetValue<int>(CommonConst.NumPartitions), 3);
var consumerService = host.Services.GetRequiredService<IConsumerService>();
var producerService = host.Services.GetRequiredService<IProducerService>();
//var kafkaOptions = host.Services.GetRequiredService<IOptions<KafkaOptionConfig>>();
//await consumerService.SubscribeAsync<object>(topic, (message) =>
//{
@ -133,23 +135,29 @@ var consumerService = host.Services.GetRequiredService<IConsumerService>();
//stopwatch.Stop();
//Console.WriteLine($"耗时: {stopwatch.ElapsedMilliseconds} 毫秒,{stopwatch.ElapsedMilliseconds/1000} 秒");
var producerService = host.Services.GetRequiredService<IProducerService>();
//int num = 840;
//while (num <= 900)
int num = 1;
while (num <= 6)
{
await producerService.ProduceAsync<TestTopic>(topic, new TestTopic { Topic = topic, Val = num });
num++;
}
//int num = 2;
//while (num <= 4)
//{
// await producerService.ProduceAsync<string>(topic, num.ToString());
// num++;
//}
//await Task.Factory.StartNew(async() => {
// int num = 0;
// while (true)
// {
// //await producerService.ProduceAsync(topic, new TestTopic { Topic = topic, Val = i });
// await producerService.ProduceAsync<string>(topic, num.ToString());
// num++;
// }
await Task.Factory.StartNew(async() => {
int num = 0;
while (true)
{
//await producerService.ProduceAsync(topic, new TestTopic { Topic = topic, Val = i });
await producerService.ProduceAsync<string>(topic, num.ToString());
num++;
}
});
//});
Console.WriteLine("\n按Esc键退出");
while (true)
{

4
modules/JiShe.CollectBus.Kafka.Test/appsettings.json
View File

@ -91,8 +91,8 @@
"SaslUserName": "lixiao",
"SaslPassword": "lixiao1980",
"KafkaReplicationFactor": 3,
"NumPartitions": 1,
"ServerTagName": "JiSheCollectBus2"
"NumPartitions": 30,
"ServerTagName": "JiSheCollectBus99"
//"Topic": {
// "ReplicationFactor": 3,
// "NumPartitions": 1000

2
modules/JiShe.CollectBus.Kafka/Attributes/KafkaSubscribeAttribute.cs
View File

@ -22,7 +22,7 @@ namespace JiShe.CollectBus.Kafka.Attributes
/// <summary>
/// 消费者组
/// </summary>
public string GroupId { get; set; } = "default";
public string? GroupId { get; set; } = null;//"default"
/// <summary>
/// 任务数(默认是多少个分区多少个任务)

12
modules/JiShe.CollectBus.Kafka/CollectBusKafkaModule.cs
View File

@ -1,6 +1,7 @@
using Confluent.Kafka;
using JiShe.CollectBus.Common.Consts;
using JiShe.CollectBus.Kafka.Consumer;
using JiShe.CollectBus.Kafka.Internal;
using JiShe.CollectBus.Kafka.Producer;
using Microsoft.AspNetCore.Builder;
using Microsoft.Extensions.Configuration;
@ -43,10 +44,19 @@ namespace JiShe.CollectBus.Kafka
//context.Services.AddHostedService<HostedService>();
}
/// <summary>
/// 在初始化之前,初始化Kafka Topic
/// </summary>
/// <param name="context"></param>
public override void OnPreApplicationInitialization(ApplicationInitializationContext context)
{
var app = context.GetApplicationBuilder();
app.ApplicationServices.UseInitKafkaTopic();
}
public override void OnApplicationInitialization(ApplicationInitializationContext context)
{
var app = context.GetApplicationBuilder();
// 注册Subscriber
app.ApplicationServices.UseKafkaSubscribe();

49
modules/JiShe.CollectBus.Kafka/Consumer/ConsumerService.cs
View File

@ -1,5 +1,7 @@
using Confluent.Kafka;
using JiShe.CollectBus.Common.Consts;
using JiShe.CollectBus.Kafka.Internal;
using JiShe.CollectBus.Kafka.Serialization;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
@ -47,11 +49,11 @@ namespace JiShe.CollectBus.Kafka.Consumer
var config = new ConsumerConfig
{
BootstrapServers = _kafkaOptionConfig.BootstrapServers,
GroupId = groupId ?? "default",
GroupId = groupId ?? _kafkaOptionConfig.ServerTagName,
AutoOffsetReset = AutoOffsetReset.Earliest,
EnableAutoCommit = false, // 禁止AutoCommit
EnablePartitionEof = true, // 启用分区末尾标记
AllowAutoCreateTopics = true, // 启用自动创建
//AllowAutoCreateTopics = true, // 启用自动创建
FetchMaxBytes = 1024 * 1024 * 50 // 增加拉取大小（50MB）
};
@ -252,7 +254,7 @@ namespace JiShe.CollectBus.Kafka.Consumer
/// <param name="groupId">消费组ID</param>
/// <param name="batchSize">批次大小</param>
/// <param name="batchTimeout">批次超时时间</param>
public async Task SubscribeBatchAsync<TKey, TValue>(string topic, Func<IEnumerable<TValue>, Task<bool>> messageBatchHandler, string? groupId = null, int batchSize = 100, TimeSpan? batchTimeout = null) where TKey : notnull where TValue : class
public async Task SubscribeBatchAsync<TKey, TValue>(string topic, Func<List<TValue>, Task<bool>> messageBatchHandler, string? groupId = null, int batchSize = 100, TimeSpan? batchTimeout = null) where TKey : notnull where TValue : class
{
await SubscribeBatchAsync<TKey, TValue>(new[] { topic }, messageBatchHandler, groupId, batchSize, batchTimeout);
}
@ -267,17 +269,17 @@ namespace JiShe.CollectBus.Kafka.Consumer
/// <param name="groupId">消费组ID</param>
/// <param name="batchSize">批次大小</param>
/// <param name="batchTimeout">批次超时时间</param>
public async Task SubscribeBatchAsync<TKey, TValue>(string[] topics,Func<IEnumerable<TValue>, Task<bool>> messageBatchHandler, string? groupId = null,int batchSize = 100, TimeSpan? batchTimeout = null) where TKey : notnull where TValue : class
public async Task SubscribeBatchAsync<TKey, TValue>(string[] topics,Func<List<TValue>, Task<bool>> messageBatchHandler, string? groupId = null,int batchSize = 100, TimeSpan? batchTimeout = null) where TKey : notnull where TValue : class
{
var consumerKey = typeof(KafkaConsumer<TKey, TValue>);
var cts = new CancellationTokenSource();
var consumer = _consumerStore.GetOrAdd(consumerKey, _ =>
(
CreateConsumer<TKey, TValue>(groupId),
cts
)).Consumer as IConsumer<TKey, TValue>;
//var consumer = _consumerStore.GetOrAdd(consumerKey, _ =>
// (
// CreateConsumer<TKey, TValue>(groupId),
// cts
// )).Consumer as IConsumer<TKey, TValue>;
var consumer = CreateConsumer<string, TValue>(groupId);
consumer!.Subscribe(topics);
var timeout = batchTimeout ?? TimeSpan.FromSeconds(5); // 默认超时时间调整为5秒
@ -300,8 +302,8 @@ namespace JiShe.CollectBus.Kafka.Consumer
{
if (result.IsPartitionEOF)
{
_logger.LogInformation("Kafka消费: {Topic} 分区 {Partition} 已消费完", result.Topic, result.Partition);
await Task.Delay(TimeSpan.FromSeconds(1), cts.Token);
//_logger.LogInformation("Kafka消费: {Topic} 分区 {Partition} 已消费完", result.Topic, result.Partition);
await Task.Delay(10, cts.Token);
}
else if (result.Message.Value != null)
{
@ -330,7 +332,7 @@ namespace JiShe.CollectBus.Kafka.Consumer
// 处理批次
if (messages.Count > 0)
{
bool success = await messageBatchHandler(messages.Select(m => m.Value));
bool success = await messageBatchHandler(messages.Select(m => m.Value).ToList());
if (success)
{
var offsetsByPartition = new Dictionary<TopicPartition, long>();
@ -383,7 +385,7 @@ namespace JiShe.CollectBus.Kafka.Consumer
/// <param name="batchSize">批次大小</param>
/// <param name="batchTimeout">批次超时时间</param>
/// <param name="consumeTimeout">消费等待时间</param>
public async Task SubscribeBatchAsync<TValue>(string topic, Func<IEnumerable<TValue>, Task<bool>> messageBatchHandler, string? groupId = null, int batchSize = 100, TimeSpan? batchTimeout = null, TimeSpan? consumeTimeout = null) where TValue : class
public async Task SubscribeBatchAsync<TValue>(string topic, Func<List<TValue>, Task<bool>> messageBatchHandler, string? groupId = null, int batchSize = 100, TimeSpan? batchTimeout = null, TimeSpan? consumeTimeout = null) where TValue : class
{
await SubscribeBatchAsync<TValue>(new[] { topic }, messageBatchHandler, groupId, batchSize, batchTimeout, consumeTimeout);
@ -400,17 +402,18 @@ namespace JiShe.CollectBus.Kafka.Consumer
/// <param name="batchSize">批次大小</param>
/// <param name="batchTimeout">批次超时时间</param>
/// <param name="consumeTimeout">消费等待时间</param>
public async Task SubscribeBatchAsync<TValue>(string[] topics,Func<IEnumerable<TValue>, Task<bool>> messageBatchHandler, string? groupId = null, int batchSize = 100,TimeSpan? batchTimeout = null,TimeSpan? consumeTimeout = null)where TValue : class
public async Task SubscribeBatchAsync<TValue>(string[] topics,Func<List<TValue>, Task<bool>> messageBatchHandler, string? groupId = null, int batchSize = 100,TimeSpan? batchTimeout = null,TimeSpan? consumeTimeout = null)where TValue : class
{
var consumerKey = typeof(KafkaConsumer<string, TValue>);
var cts = new CancellationTokenSource();
var consumer = _consumerStore.GetOrAdd(consumerKey, _ =>
(
CreateConsumer<string, TValue>(groupId),
cts
)).Consumer as IConsumer<string, TValue>;
//var consumer = _consumerStore.GetOrAdd(consumerKey, _ =>
// (
// CreateConsumer<string, TValue>(groupId),
// cts
// )).Consumer as IConsumer<string, TValue>;
var consumer= CreateConsumer<string, TValue> (groupId);
consumer!.Subscribe(topics);
var timeout = batchTimeout ?? TimeSpan.FromSeconds(5); // 默认超时时间调整为5秒
@ -434,8 +437,8 @@ namespace JiShe.CollectBus.Kafka.Consumer
{
if (result.IsPartitionEOF)
{
_logger.LogInformation("Kafka消费: {Topic} 分区 {Partition} 已消费完", result.Topic, result.Partition);
await Task.Delay(TimeSpan.FromSeconds(1), cts.Token);
//_logger.LogInformation("Kafka消费: {Topic} 分区 {Partition} 已消费完", result.Topic, result.Partition);
await Task.Delay(10, cts.Token);
}
else if (result.Message.Value != null)
{
@ -464,7 +467,7 @@ namespace JiShe.CollectBus.Kafka.Consumer
// 处理批次
if (messages.Count > 0)
{
bool success = await messageBatchHandler(messages.Select(m => m.Value));
bool success = await messageBatchHandler(messages.Select(m => m.Value).ToList());
if (success)
{
var offsetsByPartition = new Dictionary<TopicPartition, long>();

8
modules/JiShe.CollectBus.Kafka/Consumer/IConsumerService.cs
View File

@ -33,13 +33,13 @@ namespace JiShe.CollectBus.Kafka.Consumer
/// <returns></returns>
Task SubscribeAsync<TValue>(string[] topics, Func<TValue, Task<bool>> messageHandler, string? groupId = null) where TValue : class;
Task SubscribeBatchAsync<TKey, TValue>(string[] topics, Func<IEnumerable<TValue>, Task<bool>> messageBatchHandler, string? groupId = null, int batchSize = 100, TimeSpan? batchTimeout = null) where TKey : notnull where TValue : class;
Task SubscribeBatchAsync<TKey, TValue>(string[] topics, Func<List<TValue>, Task<bool>> messageBatchHandler, string? groupId = null, int batchSize = 100, TimeSpan? batchTimeout = null) where TKey : notnull where TValue : class;
Task SubscribeBatchAsync<TKey, TValue>(string topic, Func<IEnumerable<TValue>, Task<bool>> messageBatchHandler, string? groupId = null, int batchSize = 100, TimeSpan? batchTimeout = null) where TKey : notnull where TValue : class;
Task SubscribeBatchAsync<TKey, TValue>(string topic, Func<List<TValue>, Task<bool>> messageBatchHandler, string? groupId = null, int batchSize = 100, TimeSpan? batchTimeout = null) where TKey : notnull where TValue : class;
Task SubscribeBatchAsync<TValue>(string topic, Func<IEnumerable<TValue>, Task<bool>> messageBatchHandler, string? groupId = null, int batchSize = 100, TimeSpan? batchTimeout = null, TimeSpan? consumeTimeout = null) where TValue : class;
Task SubscribeBatchAsync<TValue>(string topic, Func<List<TValue>, Task<bool>> messageBatchHandler, string? groupId = null, int batchSize = 100, TimeSpan? batchTimeout = null, TimeSpan? consumeTimeout = null) where TValue : class;
Task SubscribeBatchAsync<TValue>(string[] topics, Func<IEnumerable<TValue>, Task<bool>> messageBatchHandler, string? groupId = null, int batchSize = 100, TimeSpan? batchTimeout = null, TimeSpan? consumeTimeout = null) where TValue : class;
Task SubscribeBatchAsync<TValue>(string[] topics, Func<List<TValue>, Task<bool>> messageBatchHandler, string? groupId = null, int batchSize = 100, TimeSpan? batchTimeout = null, TimeSpan? consumeTimeout = null) where TValue : class;
void Unsubscribe<TKey, TValue>() where TKey : notnull where TValue : class;
}

2
modules/JiShe.CollectBus.Kafka/HeadersFilter.cs → modules/JiShe.CollectBus.Kafka/Internal/HeadersFilter.cs
View File

@ -5,7 +5,7 @@ using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace JiShe.CollectBus.Kafka
namespace JiShe.CollectBus.Kafka.Internal
{
/// <summary>
/// 消息头过滤器

2
modules/JiShe.CollectBus.Kafka/IKafkaSubscribe.cs → modules/JiShe.CollectBus.Kafka/Internal/IKafkaSubscribe.cs
View File

@ -4,7 +4,7 @@ using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace JiShe.CollectBus.Kafka
namespace JiShe.CollectBus.Kafka.Internal
{
/// <summary>
/// Kafka订阅者

2
modules/JiShe.CollectBus.Kafka/ISubscribeAck.cs → modules/JiShe.CollectBus.Kafka/Internal/ISubscribeAck.cs
View File

@ -4,7 +4,7 @@ using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace JiShe.CollectBus.Kafka
namespace JiShe.CollectBus.Kafka.Internal
{
public interface ISubscribeAck
{

2
modules/JiShe.CollectBus.Kafka/KafkaOptionConfig.cs → modules/JiShe.CollectBus.Kafka/Internal/KafkaOptionConfig.cs
View File

@ -5,7 +5,7 @@ using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace JiShe.CollectBus.Kafka
namespace JiShe.CollectBus.Kafka.Internal
{
public class KafkaOptionConfig
{

113
modules/JiShe.CollectBus.Kafka/Internal/ReflectionHelper.cs Normal file
View File

@ -0,0 +1,113 @@
using Newtonsoft.Json;
using System;
using System.Collections;
using System.Collections.Generic;
using System.Linq;
using System.Reflection;
using System.Text;
using System.Threading.Tasks;
namespace JiShe.CollectBus.Kafka.Internal
{
/// <summary>
/// 反射辅助类
/// </summary>
public static class ReflectionHelper
{
/// <summary>
///集合类型
///Item1：参数类型
///Item2：集合元素类型
/// </summary>
public static Tuple<Type,Type?> GetParameterTypeInfo(this MethodInfo method, int parameterIndex=0)
{
// 参数校验
if (method == null) throw new ArgumentNullException(nameof(method));
var parameters = method.GetParameters();
if (parameterIndex < 0 || parameterIndex >= parameters.Length)
throw new ArgumentOutOfRangeException(nameof(parameterIndex));
ParameterInfo param = parameters[parameterIndex];
Type paramType = param.ParameterType;
Type? elementType = null;
// 判断是否是集合类型（排除字符串）
if (paramType != typeof(string) && IsEnumerableType(paramType))
{
elementType = GetEnumerableElementType(paramType);
}
return Tuple.Create(paramType, elementType);
}
/// <summary>
/// 判断是否是集合类型（排除字符串）
/// </summary>
public static bool IsEnumerableType(this Type type)
{
return type.IsArray
|| (type.IsGenericType && type.GetInterfaces()
.Any(t => t.IsGenericType
&& t.GetGenericTypeDefinition() == typeof(IEnumerable<>)))
|| type.GetInterfaces().Any(t => t == typeof(System.Collections.IEnumerable));
}
/// <summary>
/// 获取集合元素的类型
/// </summary>
public static Type? GetEnumerableElementType(this Type type)
{
// 处理数组类型
if (type.IsArray)
return type.GetElementType();
// 处理直接实现IEnumerable<T>的类型（如IEnumerable<int>本身）
if (type.IsGenericType && type.GetGenericTypeDefinition() == typeof(IEnumerable<>))
return type.GetGenericArguments()[0];
// 处理通过接口实现IEnumerable<T>的泛型集合（如List<T>）
var genericEnumerable = type.GetInterfaces()
.FirstOrDefault(t => t.IsGenericType
&& t.GetGenericTypeDefinition() == typeof(IEnumerable<>));
if (genericEnumerable != null)
return genericEnumerable.GetGenericArguments()[0];
// 处理非泛型集合类型（如 ArrayList）
if (typeof(IEnumerable).IsAssignableFrom(type) && type == typeof(ArrayList))
return typeof(ArrayList);
// 返回null表示无法确定元素类型
return null;
}
// <summary>
/// 判断是否使用强转换
/// </summary>
/// <param name="targetType">目标类型</param>
/// <returns></returns>
public static bool IsConvertType(this Type targetType)
{
// 处理可空类型
Type underlyingType = Nullable.GetUnderlyingType(targetType) ?? targetType;
// 情况1：值类型或基元类型（如 int、DateTime）
if (underlyingType.IsValueType || underlyingType.IsPrimitive)
return true;
// 情况2：字符串类型直接赋值
else if (underlyingType == typeof(string))
return true;
// 情况3：枚举类型处理
//else if (underlyingType.IsEnum)
//{
// if (Enum.IsDefined(underlyingType, msg))
// {
// convertedValue = Enum.Parse(underlyingType, msg.ToString());
// return true;
// }
// return false;
//}
return false;
}
}
}

2
modules/JiShe.CollectBus.Kafka/SubscribeResult.cs → modules/JiShe.CollectBus.Kafka/Internal/SubscribeResult.cs
View File

@ -6,7 +6,7 @@ using System.Text;
using System.Threading.Tasks;
using static System.Runtime.InteropServices.JavaScript.JSType;
namespace JiShe.CollectBus.Kafka
namespace JiShe.CollectBus.Kafka.Internal
{
public class SubscribeResult: ISubscribeAck
{

156
modules/JiShe.CollectBus.Kafka/KafkaSubcribesExtensions.cs
View File

@ -5,30 +5,33 @@ using JiShe.CollectBus.Common.Helpers;
using JiShe.CollectBus.Kafka.AdminClient;
using JiShe.CollectBus.Kafka.Attributes;
using JiShe.CollectBus.Kafka.Consumer;
using JiShe.CollectBus.Kafka.Internal;
using JiShe.CollectBus.Kafka.Serialization;
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Mvc.Abstractions;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using Newtonsoft.Json;
using System;
using System.Collections;
using System.Collections.Generic;
using System.ComponentModel;
using System.Linq.Expressions;
using System.Reflection;
using System.Text.Json;
using System.Threading.Tasks;
using YamlDotNet.Core.Tokens;
using static System.Runtime.InteropServices.JavaScript.JSType;
namespace JiShe.CollectBus.Kafka
{
public static class KafkaSubcribesExtensions
{
/// <summary>
/// 添加Kafka订阅
/// </summary>
/// <param name="app"></param>
/// <param name="assembly"></param>
public static void UseKafkaSubscribe(this IServiceProvider provider)
{
var lifetime = provider.GetRequiredService<IHostApplicationLifetime>();
public static void UseInitKafkaTopic(this IServiceProvider provider)
{
//初始化主题信息
var kafkaAdminClient = provider.GetRequiredService<IAdminClientService>();
var kafkaOptions = provider.GetRequiredService<IOptions<KafkaOptionConfig>>();
@ -40,6 +43,17 @@ namespace JiShe.CollectBus.Kafka
{
kafkaAdminClient.CreateTopicAsync(item, kafkaOptions.Value.NumPartitions, kafkaOptions.Value.KafkaReplicationFactor).ConfigureAwait(false).GetAwaiter().GetResult();
}
}
/// <summary>
/// 添加Kafka订阅
/// </summary>
/// <param name="app"></param>
/// <param name="assembly"></param>
public static void UseKafkaSubscribe(this IServiceProvider provider)
{
var lifetime = provider.GetRequiredService<IHostApplicationLifetime>();
var kafkaOptions = provider.GetRequiredService<IOptions<KafkaOptionConfig>>();
lifetime.ApplicationStarted.Register(() =>
{
var logger = provider.GetRequiredService<ILogger<CollectBusKafkaModule>>();
@ -88,17 +102,7 @@ namespace JiShe.CollectBus.Kafka
{
var provider = app.ApplicationServices;
var lifetime = provider.GetRequiredService<IHostApplicationLifetime>();
//初始化主题信息
var kafkaAdminClient = provider.GetRequiredService<IAdminClientService>();
var kafkaOptions = provider.GetRequiredService<IOptions<KafkaOptionConfig>>();
List<string> topics = ProtocolConstExtensions.GetAllTopicNamesByIssued();
topics.AddRange(ProtocolConstExtensions.GetAllTopicNamesByReceived());
foreach (var item in topics)
{
kafkaAdminClient.CreateTopicAsync(item, kafkaOptions.Value.NumPartitions, kafkaOptions.Value.KafkaReplicationFactor).ConfigureAwait(false).GetAwaiter().GetResult();
}
lifetime.ApplicationStarted.Register(() =>
{
var logger = provider.GetRequiredService<ILogger<CollectBusKafkaModule>>();
@ -176,12 +180,12 @@ namespace JiShe.CollectBus.Kafka
if (attr.EnableBatch)
{
await consumerService.SubscribeBatchAsync<object>(attr.Topic, async (message) =>
await consumerService.SubscribeBatchAsync<dynamic>(attr.Topic, async (message) =>
{
try
{
#if DEBUG
logger.LogInformation($"kafka批量消费消息:{message}");
logger.LogInformation($"kafka批量消费消息:{message.Serialize()}");
#endif
// 处理消息
return await ProcessMessageAsync(message.ToList(), method, subscribe);
@ -196,7 +200,7 @@ namespace JiShe.CollectBus.Kafka
}
else
{
await consumerService.SubscribeAsync<object>(attr.Topic, async (message) =>
await consumerService.SubscribeAsync<dynamic>(attr.Topic, async (message) =>
{
try
{
@ -225,26 +229,112 @@ namespace JiShe.CollectBus.Kafka
/// <param name="method"></param>
/// <param name="subscribe"></param>
/// <returns></returns>
private static async Task<bool> ProcessMessageAsync(List<object> messages, MethodInfo method, object subscribe)
private static async Task<bool> ProcessMessageAsync(List<dynamic> messages, MethodInfo method, object subscribe)
{
var parameters = method.GetParameters();
bool isGenericTask = method.ReturnType.IsGenericType
&& method.ReturnType.GetGenericTypeDefinition() == typeof(Task<>);
bool existParameters = parameters.Length > 0;
List<object>? messageObj = null;
object[]? executeParameters = null;
if (existParameters)
{
messageObj = new List<object>();
var paramType = parameters[0].ParameterType;
foreach (var msg in messages)
IList? list = null;
Tuple<Type, Type?> tuple = method.GetParameterTypeInfo();
bool isEnumerable = false;
if (tuple.Item2 != null)
{
var data = paramType != typeof(string) ? msg?.ToString()?.Deserialize(paramType) : msg;
if (data != null)
messageObj.Add(data);
Type listType = typeof(List<>).MakeGenericType(tuple.Item2);
list = (IList)Activator.CreateInstance(listType)!;
isEnumerable = tuple.Item2.IsConvertType();
}
else
{
isEnumerable = tuple.Item1.IsConvertType();
}
#region 暂时
//foreach (var msg in messages)
//{
// if (tuple.Item2 != null)
// {
// if (isEnumerable)
// {
// var parameterType = parameters[0].ParameterType;
// var data=messages?.Serialize().Deserialize(parameterType);
// messageObj = data!=null? new[] { data }:null;
// break;
// }
// else
// {
// // 集合类型
// var data = msg?.Serialize().Deserialize(tuple.Item2) /*isEnumerable ? Convert.ChangeType(msg, tuple.Item2) : msg?.Serialize().Deserialize(tuple.Item2)*/;
// if (data != null)
// list?.Add(data);
// }
// }
// else
// {
// // (dynamic)Convert.ChangeType(msg, tuple.Item1)
// using (var stream = new MemoryStream(msg))
// {
// var data1= System.Text.Json.JsonSerializer.Deserialize(stream, tuple.Item1);
// }
// var data = isEnumerable ? System.Text.Json.JsonSerializer.Deserialize(msg, tuple.Item1): msg?.ToString()?.Deserialize(tuple.Item1);
// if (data != null)
// messageObj = new[] { data };
// }
//}
//if (tuple.Item2 != null && list != null && list.Count > 0)
//{
// messageObj = new[] { list };
//}
#endregion
var parameterDescriptors = method.GetParameters();
executeParameters = new object?[parameterDescriptors.Length];
for (var i = 0; i < parameterDescriptors.Length; i++)
{
foreach (var item in messages)
{
object? tempParameter=null;
var parameterDescriptor = parameterDescriptors[i];
if (KafkaSerialization.IsJsonType(item))
{
tempParameter = KafkaSerialization.Deserialize(item, tuple.Item2 != null? tuple.Item2: parameterDescriptor.ParameterType);
}
else
{
var converter = TypeDescriptor.GetConverter(parameterDescriptor.ParameterType);
if (converter.CanConvertFrom(item.GetType()))
{
tempParameter = converter.ConvertFrom(item);
}
else
{
if (parameterDescriptor.ParameterType.IsInstanceOfType(item))
tempParameter = item;
else
tempParameter =Convert.ChangeType(item, parameterDescriptor.ParameterType);
}
}
if (tuple.Item2 == null)
{
executeParameters[i] = tempParameter;
}
else
{
list.Add(tempParameter);
}
}
if(list!=null && list.Count>0)
executeParameters[i] = list;
}
}
var result = method.Invoke(subscribe, messageObj?.ToArray());
var result = method.Invoke(subscribe, executeParameters);
if (result is Task<ISubscribeAck> genericTask)
{
await genericTask.ConfigureAwait(false);
@ -262,5 +352,9 @@ namespace JiShe.CollectBus.Kafka
return false;
}
}
}

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

@ -6,6 +6,8 @@ using System.Text;
using System.Threading.Tasks;
using Confluent.Kafka;
using JiShe.CollectBus.Kafka.Consumer;
using JiShe.CollectBus.Kafka.Internal;
using JiShe.CollectBus.Kafka.Serialization;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
@ -56,7 +58,7 @@ namespace JiShe.CollectBus.Kafka.Producer
var config = new ProducerConfig
{
BootstrapServers = _kafkaOptionConfig.BootstrapServers,
AllowAutoCreateTopics = true,
//AllowAutoCreateTopics = true,
QueueBufferingMaxKbytes = 2_097_151, // 修改缓冲区最大为2GB，默认为1GB
CompressionType = CompressionType.Lz4, // 配置使用压缩算法LZ4，其他：gzip/snappy/zstd
BatchSize = 32_768, // 修改批次大小为32K

43
modules/JiShe.CollectBus.Kafka/JsonSerializer.cs → modules/JiShe.CollectBus.Kafka/Serialization/JsonSerializer.cs
View File

@ -8,7 +8,7 @@ using Confluent.Kafka;
using System.Text.Json.Serialization;
using System.Text.Encodings.Web;
namespace JiShe.CollectBus.Kafka
namespace JiShe.CollectBus.Kafka.Serialization
{
/// <summary>
/// JSON 序列化器（支持泛型）
@ -49,10 +49,11 @@ namespace JiShe.CollectBus.Kafka
{
if (isNull)
return default;
try
{
return JsonSerializer.Deserialize<T>(data, _options);
if (data.IsEmpty)
return default;
return JsonSerializer.Deserialize<T>(data, _options)!;
}
catch (Exception ex)
{
@ -85,4 +86,40 @@ namespace JiShe.CollectBus.Kafka
writer.WriteStringValue(value.ToString(_dateFormatString));
}
}
public static class KafkaSerialization
{
/// <summary>
/// 判断是否是json类型
/// </summary>
/// <param name="jsonObject"></param>
/// <returns></returns>
public static bool IsJsonType(this object jsonObject)
{
return jsonObject is JsonElement;
}
public static object? Deserialize(object value, Type valueType)
{
var _jsonSerializerOptions = new JsonSerializerOptions
{
DefaultIgnoreCondition = JsonIgnoreCondition.Never,
WriteIndented = false,// 设置格式化输出
Encoder = JavaScriptEncoder.UnsafeRelaxedJsonEscaping,// 允许特殊字符
IgnoreReadOnlyFields = true,
IgnoreReadOnlyProperties = true,
NumberHandling = JsonNumberHandling.AllowReadingFromString, // 允许数字字符串
AllowTrailingCommas = true, // 忽略尾随逗号
ReadCommentHandling = JsonCommentHandling.Skip, // 忽略注释
PropertyNameCaseInsensitive = true, // 属性名称大小写不敏感
PropertyNamingPolicy = JsonNamingPolicy.CamelCase, // 属性名称使用驼峰命名规则
Converters = { new DateTimeJsonConverter() } // 注册你的自定义转换器,
};
if (value is JsonElement jsonElement) return jsonElement.Deserialize(valueType, _jsonSerializerOptions);
throw new NotSupportedException("Type is not of type JsonElement");
}
}
}

13
services/JiShe.CollectBus.Application.Contracts/Subscribers/ISubscriberAppService.cs
View File

@ -1,17 +1,18 @@
using System.Threading.Tasks;
using System.Collections.Generic;
using System.Threading.Tasks;
using JiShe.CollectBus.Common.Models;
using JiShe.CollectBus.IotSystems.MessageReceiveds;
using JiShe.CollectBus.Kafka;
using JiShe.CollectBus.Kafka.Internal;
using Volo.Abp.Application.Services;
namespace JiShe.CollectBus.Subscribers
{
public interface ISubscriberAppService : IApplicationService
{
Task<ISubscribeAck> LoginIssuedEvent(IssuedEventMessage issuedEventMessage);
Task<ISubscribeAck> HeartbeatIssuedEvent(IssuedEventMessage issuedEventMessage);
Task<ISubscribeAck> LoginIssuedEvent(List<IssuedEventMessage> issuedEventMessage);
Task<ISubscribeAck> HeartbeatIssuedEvent(List<IssuedEventMessage> issuedEventMessage);
Task<ISubscribeAck> ReceivedEvent(MessageReceived receivedMessage);
Task<ISubscribeAck> ReceivedHeartbeatEvent(MessageReceivedHeartbeat receivedHeartbeatMessage);
Task<ISubscribeAck> ReceivedLoginEvent(MessageReceivedLogin receivedLoginMessage);
Task<ISubscribeAck> ReceivedHeartbeatEvent(List<MessageReceivedHeartbeat> receivedHeartbeatMessage);
Task<ISubscribeAck> ReceivedLoginEvent(List<MessageReceivedLogin> receivedLoginMessage);
}
}

2
services/JiShe.CollectBus.Application.Contracts/Subscribers/IWorkerSubscriberAppService.cs
View File

@ -1,7 +1,7 @@
using JiShe.CollectBus.IotSystems.MessageIssueds;
using JiShe.CollectBus.IotSystems.MessageReceiveds;
using JiShe.CollectBus.IotSystems.MeterReadingRecords;
using JiShe.CollectBus.Kafka;
using JiShe.CollectBus.Kafka.Internal;
using System.Collections.Generic;
using System.Threading.Tasks;
using Volo.Abp.Application.Services;

1
services/JiShe.CollectBus.Application/CollectBusApplicationModule.cs
View File

@ -23,6 +23,7 @@ using Volo.Abp.BackgroundWorkers.Hangfire;
using Volo.Abp.EventBus;
using Volo.Abp.Modularity;
using Microsoft.Extensions.Options;
using JiShe.CollectBus.Kafka.Internal;
namespace JiShe.CollectBus;

2
services/JiShe.CollectBus.Application/Samples/SampleAppService.cs
View File

@ -20,13 +20,13 @@ using System.Diagnostics.Metrics;
using JiShe.CollectBus.Common.DeviceBalanceControl;
using JiShe.CollectBus.Kafka.Attributes;
using System.Text.Json;
using JiShe.CollectBus.Kafka;
using JiShe.CollectBus.Application.Contracts;
using JiShe.CollectBus.Common.Models;
using System.Diagnostics;
using JiShe.CollectBus.IoTDB.Context;
using JiShe.CollectBus.IoTDB.Interface;
using JiShe.CollectBus.IoTDB.Options;
using JiShe.CollectBus.Kafka.Internal;
namespace JiShe.CollectBus.Samples;

2
services/JiShe.CollectBus.Application/ScheduledMeterReading/BasicScheduledMeterReadingService.cs
View File

@ -11,7 +11,6 @@ using JiShe.CollectBus.GatherItem;
using JiShe.CollectBus.IotSystems.MessageIssueds;
using JiShe.CollectBus.IotSystems.MeterReadingRecords;
using JiShe.CollectBus.IotSystems.Watermeter;
using JiShe.CollectBus.Kafka;
using JiShe.CollectBus.Kafka.Producer;
using JiShe.CollectBus.Protocol.Contracts;
using JiShe.CollectBus.RedisDataCache;
@ -27,6 +26,7 @@ using System.Threading;
using System.Threading.Tasks;
using JiShe.CollectBus.IoTDB.Interface;
using static FreeSql.Internal.GlobalFilter;
using JiShe.CollectBus.Kafka.Internal;
namespace JiShe.CollectBus.ScheduledMeterReading
{

2
services/JiShe.CollectBus.Application/ScheduledMeterReading/EnergySystemScheduledMeterReadingService.cs
View File

@ -13,7 +13,7 @@ using JiShe.CollectBus.IotSystems.Devices;
using JiShe.CollectBus.IotSystems.MessageIssueds;
using JiShe.CollectBus.IotSystems.MeterReadingRecords;
using JiShe.CollectBus.IotSystems.Watermeter;
using JiShe.CollectBus.Kafka;
using JiShe.CollectBus.Kafka.Internal;
using JiShe.CollectBus.Kafka.Producer;
using JiShe.CollectBus.Repository;
using JiShe.CollectBus.Repository.MeterReadingRecord;

34
services/JiShe.CollectBus.Application/Subscribers/SubscriberAppService.cs
View File

@ -5,7 +5,6 @@ using JiShe.CollectBus.Common.Models;
using JiShe.CollectBus.IotSystems.Devices;
using JiShe.CollectBus.IotSystems.MessageReceiveds;
using JiShe.CollectBus.IotSystems.MeterReadingRecords;
using JiShe.CollectBus.Kafka;
using JiShe.CollectBus.Kafka.Attributes;
using JiShe.CollectBus.Protocol.Contracts;
using JiShe.CollectBus.Protocol.Contracts.Interfaces;
@ -19,6 +18,8 @@ using System.Threading.Tasks;
using JiShe.CollectBus.IoTDB.Interface;
using TouchSocket.Sockets;
using Volo.Abp.Domain.Repositories;
using System.Collections.Generic;
using JiShe.CollectBus.Kafka.Internal;
namespace JiShe.CollectBus.Subscribers
{
@ -65,11 +66,13 @@ namespace JiShe.CollectBus.Subscribers
_dbProvider = dbProvider;
}
[KafkaSubscribe(ProtocolConst.SubscriberLoginIssuedEventName)]
[KafkaSubscribe(ProtocolConst.SubscriberLoginIssuedEventName, EnableBatch = true)]
//[CapSubscribe(ProtocolConst.SubscriberLoginIssuedEventName)]
public async Task<ISubscribeAck> LoginIssuedEvent(IssuedEventMessage issuedEventMessage)
public async Task<ISubscribeAck> LoginIssuedEvent(List<IssuedEventMessage> issuedEventMessages)
{
bool isAck = false;
foreach (var issuedEventMessage in issuedEventMessages)
{
switch (issuedEventMessage.Type)
{
case IssuedEventType.Heartbeat:
@ -95,14 +98,18 @@ namespace JiShe.CollectBus.Subscribers
//}
await _tcpService.SendAsync(issuedEventMessage.ClientId, issuedEventMessage.Message);
}
return isAck? SubscribeAck.Success(): SubscribeAck.Fail();
}
[KafkaSubscribe(ProtocolConst.SubscriberHeartbeatIssuedEventName)]
[KafkaSubscribe(ProtocolConst.SubscriberHeartbeatIssuedEventName, EnableBatch = true)]
//[CapSubscribe(ProtocolConst.SubscriberHeartbeatIssuedEventName)]
public async Task<ISubscribeAck> HeartbeatIssuedEvent(IssuedEventMessage issuedEventMessage)
public async Task<ISubscribeAck> HeartbeatIssuedEvent(List<IssuedEventMessage> issuedEventMessages)
{
bool isAck = false;
foreach (var issuedEventMessage in issuedEventMessages)
{
switch (issuedEventMessage.Type)
{
case IssuedEventType.Heartbeat:
@ -126,6 +133,8 @@ namespace JiShe.CollectBus.Subscribers
//}
await _tcpService.SendAsync(issuedEventMessage.ClientId, issuedEventMessage.Message);
}
return isAck ? SubscribeAck.Success() : SubscribeAck.Fail();
}
@ -186,9 +195,11 @@ namespace JiShe.CollectBus.Subscribers
return SubscribeAck.Success();
}
[KafkaSubscribe(ProtocolConst.SubscriberHeartbeatReceivedEventName)]
[KafkaSubscribe(ProtocolConst.SubscriberHeartbeatReceivedEventName, EnableBatch = true)]
//[CapSubscribe(ProtocolConst.SubscriberHeartbeatReceivedEventName)]
public async Task<ISubscribeAck> ReceivedHeartbeatEvent(MessageReceivedHeartbeat receivedHeartbeatMessage)
public async Task<ISubscribeAck> ReceivedHeartbeatEvent(List<MessageReceivedHeartbeat> receivedHeartbeatMessages)
{
foreach (var receivedHeartbeatMessage in receivedHeartbeatMessages)
{
var protocolPlugin = _serviceProvider.GetKeyedService<IProtocolPlugin>("StandardProtocolPlugin");
if (protocolPlugin == null)
@ -200,12 +211,15 @@ namespace JiShe.CollectBus.Subscribers
await protocolPlugin.HeartbeatAsync(receivedHeartbeatMessage);
await _messageReceivedHeartbeatEventRepository.InsertAsync(receivedHeartbeatMessage);
}
}
return SubscribeAck.Success();
}
[KafkaSubscribe(ProtocolConst.SubscriberLoginReceivedEventName)]
[KafkaSubscribe(ProtocolConst.SubscriberLoginReceivedEventName,EnableBatch =true)]
//[CapSubscribe(ProtocolConst.SubscriberLoginReceivedEventName)]
public async Task<ISubscribeAck> ReceivedLoginEvent(MessageReceivedLogin receivedLoginMessage)
public async Task<ISubscribeAck> ReceivedLoginEvent(List<MessageReceivedLogin> receivedLoginMessages)
{
foreach (var receivedLoginMessage in receivedLoginMessages)
{
var protocolPlugin = _serviceProvider.GetKeyedService<IProtocolPlugin>("StandardProtocolPlugin");
if (protocolPlugin == null)
@ -217,6 +231,8 @@ namespace JiShe.CollectBus.Subscribers
await protocolPlugin.LoginAsync(receivedLoginMessage);
await _messageReceivedLoginEventRepository.InsertAsync(receivedLoginMessage);
}
}
return SubscribeAck.Success();
}
}

2
services/JiShe.CollectBus.Application/Subscribers/WorkerSubscriberAppService.cs
View File

@ -8,8 +8,8 @@ using JiShe.CollectBus.IotSystems.Devices;
using JiShe.CollectBus.IotSystems.MessageIssueds;
using JiShe.CollectBus.IotSystems.MessageReceiveds;
using JiShe.CollectBus.IotSystems.MeterReadingRecords;
using JiShe.CollectBus.Kafka;
using JiShe.CollectBus.Kafka.Attributes;
using JiShe.CollectBus.Kafka.Internal;
using JiShe.CollectBus.Protocol.Contracts;
using JiShe.CollectBus.Protocol.Contracts.Interfaces;
using JiShe.CollectBus.Repository.MeterReadingRecord;

6
web/JiShe.CollectBus.Host/appsettings.json
View File

@ -43,7 +43,7 @@
"Configuration": "192.168.1.9:6380,password=1q2w3e!@#,syncTimeout=30000,abortConnect=false,connectTimeout=30000,allowAdmin=true",
"MaxPoolSize": "50",
"DefaultDB": "14",
"HangfireDB": "15"
"HangfireDB": "13"
},
"Jwt": {
"Audience": "JiShe.CollectBus",
@ -84,7 +84,7 @@
"SaslPassword": "lixiao1980",
"KafkaReplicationFactor": 3,
"NumPartitions": 30,
"ServerTagName": "JiSheCollectBus2"
"ServerTagName": "JiSheCollectBus99"
},
"IoTDBOptions": {
"UserName": "root",
@ -95,7 +95,7 @@
"OpenDebugMode": true,
"UseTableSessionPoolByDefault": false
},
"ServerTagName": "JiSheCollectBus3",
"ServerTagName": "JiSheCollectBus",
"Cassandra": {
"ReplicationStrategy": {
"Class": "NetworkTopologyStrategy", //策略为NetworkTopologyStrategy时才会有多个数据中心，SimpleStrategy用在只有一个数据中心的情况下