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

优化订阅回调

Browse Source
This commit is contained in:
zenghongyao 2025-04-15 18:03:51 +08:00
parent 7b1f04a452
commit 72d1b9f623
3 changed files with 28 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
src/JiShe.CollectBus.Application/Samples/SampleAppService.cs
View File

@ -215,11 +215,19 @@ public class SampleAppService : CollectBusAppService, ISampleAppService, IKafkaS
return aa == null;
}
[KafkaSubscribe(["test-topic"])]
[KafkaSubscribe(["test-topic1"])]
public async Task<ISubscribeAck> KafkaSubscribeAsync(object obj)
public async Task<ISubscribeAck> KafkaSubscribeAsync() // TestSubscribe obj
{
var obj=string.Empty;
_logger.LogWarning($"收到订阅消息： {obj}");
return SubscribeAck.Success();
}
}
public class TestSubscribe
{
public string Topic { get; set; }
public int Val { get; set; }
}

1
src/JiShe.CollectBus.Host/Pages/Monitor.cshtml
View File

@ -16,6 +16,7 @@
<meta name="viewport" content="width=device-width, initial-scale=1.0"/>
<link href="libs/bootstrap/css/bootstrap.min.css" rel="stylesheet"/>
<title>后端服务</title>
</head>
<body>

27
src/JiShe.CollectBus.KafkaProducer/KafkaSubcribesExtensions.cs
View File

@ -1,4 +1,7 @@
using Confluent.Kafka;
using DeviceDetectorNET;
using JiShe.CollectBus.Common.Enums;
using JiShe.CollectBus.Common.Helpers;
using JiShe.CollectBus.Kafka.Attributes;
using JiShe.CollectBus.Kafka.Consumer;
using Microsoft.AspNetCore.Builder;
@ -8,6 +11,7 @@ using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Primitives;
using Newtonsoft.Json;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Reflection;
@ -96,6 +100,7 @@ namespace JiShe.CollectBus.Kafka
});
}
/// <summary>
/// 处理消息
/// </summary>
@ -106,15 +111,18 @@ namespace JiShe.CollectBus.Kafka
private static async Task<bool> ProcessMessageAsync(string message, MethodInfo method, object subscribe)
{
var parameters = method.GetParameters();
if (parameters.Length != 1)
return true;
var paramType = parameters[0].ParameterType;
var messageObj = paramType == typeof(string)? message: JsonConvert.DeserializeObject(message, paramType);
if (method.ReturnType == typeof(Task))
bool isGenericTask = method.ReturnType.IsGenericType
&& method.ReturnType.GetGenericTypeDefinition() == typeof(Task<>);
bool existParameters = parameters.Length > 0;
dynamic? messageObj= null;
if (existParameters)
{
object? result = await (Task<bool>)method.Invoke(subscribe, new[] { messageObj })!;
var paramType = parameters[0].ParameterType;
messageObj = paramType == typeof(string) ? message : message.Deserialize(paramType);
}
if (isGenericTask)
{
object? result = await (Task<ISubscribeAck>)method.Invoke(subscribe, existParameters? new[] { messageObj }:null)!;
if (result is ISubscribeAck ackResult)
{
return ackResult.Ack;
@ -122,13 +130,12 @@ namespace JiShe.CollectBus.Kafka
}
else
{
object? result = method.Invoke(subscribe, new[] { messageObj });
object? result = method.Invoke(subscribe, existParameters ? new[] { messageObj } : null);
if (result is ISubscribeAck ackResult)
{
return ackResult.Ack;
}
}
return false;
}