Compare commits
No commits in common. "86da556e9840fa1a36c9e3dc5e21161caad77433" and "54c1643015d69084ad5739abaeb7e1ed27df1f90" have entirely different histories.
86da556e98
...
54c1643015
@ -79,7 +79,7 @@ public class AdminClientService : IAdminClientService, IDisposable, ISingletonDe
|
||||
public async Task<List<string>> ListTopicsAsync()
|
||||
{
|
||||
var metadata = Instance.GetMetadata(TimeSpan.FromSeconds(10));
|
||||
return await Task.FromResult(new List<string>(metadata.Topics.Select(t => t.Topic)));
|
||||
return new List<string>(metadata.Topics.Select(t => t.Topic));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@ -90,7 +90,7 @@ public class AdminClientService : IAdminClientService, IDisposable, ISingletonDe
|
||||
public async Task<bool> TopicExistsAsync(string topic)
|
||||
{
|
||||
var metadata = Instance.GetMetadata(TimeSpan.FromSeconds(10));
|
||||
return await Task.FromResult(metadata.Topics.Any(t => t.Topic == topic));
|
||||
return metadata.Topics.Any(t => t.Topic == topic);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
|
||||
@ -12,7 +12,6 @@ using System.Collections.Concurrent;
|
||||
using System.Text;
|
||||
using System.Text.RegularExpressions;
|
||||
using System.Threading;
|
||||
using YamlDotNet.Core.Tokens;
|
||||
|
||||
namespace JiShe.CollectBus.Kafka.Consumer
|
||||
{
|
||||
@ -133,21 +132,18 @@ namespace JiShe.CollectBus.Kafka.Consumer
|
||||
{
|
||||
await _kafkaPollyPipeline.KafkaPipeline.ExecuteAsync(async token =>
|
||||
{
|
||||
|
||||
var consumerKey = $"{groupId}_{string.Join("_", topics)}_{typeof(TKey).Name}_{typeof(TValue).Name}";
|
||||
var consumerStore = _consumerStore.GetOrAdd(consumerKey, _ =>
|
||||
var cts = new CancellationTokenSource();
|
||||
|
||||
var consumer = _consumerStore.GetOrAdd(consumerKey, _ =>
|
||||
(
|
||||
CreateConsumer<TKey, TValue>(groupId),
|
||||
new CancellationTokenSource()
|
||||
));
|
||||
if (consumerStore.Consumer == null)
|
||||
{
|
||||
_logger.LogWarning($"{string.Join("、", topics)}创建消息消费失败或消费组已被释放");
|
||||
return;
|
||||
}
|
||||
var consumer = consumerStore.Consumer as IConsumer<TKey, TValue>;
|
||||
var cts = consumerStore.CTS;
|
||||
cts
|
||||
)).Consumer as IConsumer<TKey, TValue>;
|
||||
|
||||
consumer!.Subscribe(topics);
|
||||
|
||||
_ = Task.Run(async () =>
|
||||
{
|
||||
while (!cts.IsCancellationRequested)
|
||||
@ -199,11 +195,6 @@ namespace JiShe.CollectBus.Kafka.Consumer
|
||||
{
|
||||
//ignore
|
||||
}
|
||||
catch (ObjectDisposedException)
|
||||
{
|
||||
_logger.LogError($"{string.Join("、", topics)}消费者被释放");
|
||||
break;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "处理消息时发生未知错误");
|
||||
@ -238,19 +229,12 @@ namespace JiShe.CollectBus.Kafka.Consumer
|
||||
await _kafkaPollyPipeline.KafkaPipeline.ExecuteAsync(async token =>
|
||||
{
|
||||
var consumerKey = $"{groupId}_{string.Join("_", topics)}_{typeof(Ignore).Name}_{typeof(TValue).Name}";
|
||||
|
||||
var consumerStore = _consumerStore.GetOrAdd(consumerKey, _ =>
|
||||
var cts = new CancellationTokenSource();
|
||||
var consumer = _consumerStore.GetOrAdd(consumerKey, _ =>
|
||||
(
|
||||
CreateConsumer<Ignore, TValue>(groupId),
|
||||
new CancellationTokenSource()
|
||||
));
|
||||
if (consumerStore.Consumer == null)
|
||||
{
|
||||
_logger.LogWarning($"{string.Join("、", topics)}创建消息消费失败或消费组已被释放");
|
||||
return;
|
||||
}
|
||||
var consumer = consumerStore.Consumer as IConsumer<Ignore, TValue>;
|
||||
var cts = consumerStore.CTS;
|
||||
cts
|
||||
)).Consumer as IConsumer<Ignore, TValue>;
|
||||
|
||||
consumer!.Subscribe(topics);
|
||||
|
||||
@ -304,11 +288,6 @@ namespace JiShe.CollectBus.Kafka.Consumer
|
||||
{
|
||||
//ignore
|
||||
}
|
||||
catch (ObjectDisposedException)
|
||||
{
|
||||
_logger.LogError($"{string.Join("、", topics)}消费者被释放");
|
||||
break;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "处理消息时发生未知错误");
|
||||
@ -367,20 +346,13 @@ namespace JiShe.CollectBus.Kafka.Consumer
|
||||
{
|
||||
|
||||
var consumerKey = $"{groupId}_{string.Join("_", topics)}_{typeof(TKey).Name}_{typeof(TValue).Name}";
|
||||
var cts = new CancellationTokenSource();
|
||||
|
||||
var consumerStore = _consumerStore.GetOrAdd(consumerKey, _ =>
|
||||
(
|
||||
CreateConsumer<TKey, TValue>(groupId),
|
||||
new CancellationTokenSource()
|
||||
));
|
||||
if (consumerStore.Consumer == null)
|
||||
{
|
||||
_logger.LogWarning($"{string.Join("、", topics)}创建消息消费失败或消费组已被释放");
|
||||
return;
|
||||
}
|
||||
var consumer = consumerStore.Consumer as IConsumer<TKey, TValue>;
|
||||
var cts = consumerStore.CTS;
|
||||
|
||||
var consumer = _consumerStore.GetOrAdd(consumerKey, _ =>
|
||||
(
|
||||
CreateConsumer<TKey, TValue>(groupId),
|
||||
cts
|
||||
)).Consumer as IConsumer<TKey, TValue>;
|
||||
consumer!.Subscribe(topics);
|
||||
|
||||
var timeout = batchTimeout ?? TimeSpan.FromSeconds(5); // 默认超时时间调整为5秒
|
||||
@ -472,11 +444,6 @@ namespace JiShe.CollectBus.Kafka.Consumer
|
||||
{
|
||||
//ignore
|
||||
}
|
||||
catch (ObjectDisposedException)
|
||||
{
|
||||
_logger.LogError($"{string.Join("、", topics)}消费者被释放");
|
||||
break;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "处理批量消息时发生未知错误");
|
||||
@ -538,18 +505,13 @@ namespace JiShe.CollectBus.Kafka.Consumer
|
||||
{
|
||||
|
||||
var consumerKey = $"{groupId}_{string.Join("_", topics)}_{typeof(Ignore).Name}_{typeof(TValue).Name}";
|
||||
var consumerStore = _consumerStore.GetOrAdd(consumerKey, _ =>
|
||||
(
|
||||
CreateConsumer<Ignore, TValue>(groupId),
|
||||
new CancellationTokenSource()
|
||||
));
|
||||
if (consumerStore.Consumer == null)
|
||||
{
|
||||
_logger.LogWarning($"{string.Join("、", topics)}创建消息消费失败或消费组已被释放");
|
||||
return;
|
||||
}
|
||||
var consumer = consumerStore.Consumer as IConsumer<Ignore, TValue>;
|
||||
var cts = consumerStore.CTS;
|
||||
var cts = new CancellationTokenSource();
|
||||
|
||||
var consumer = _consumerStore.GetOrAdd(consumerKey, _ =>
|
||||
(
|
||||
CreateConsumer<Ignore, TValue>(groupId),
|
||||
cts
|
||||
)).Consumer as IConsumer<Ignore, TValue>;
|
||||
|
||||
consumer!.Subscribe(topics);
|
||||
|
||||
@ -640,11 +602,6 @@ namespace JiShe.CollectBus.Kafka.Consumer
|
||||
{
|
||||
//ignore
|
||||
}
|
||||
catch (ObjectDisposedException)
|
||||
{
|
||||
_logger.LogError($"{string.Join("、", topics)}消费者被释放");
|
||||
break;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "处理批量消息时发生未知错误");
|
||||
@ -673,13 +630,11 @@ namespace JiShe.CollectBus.Kafka.Consumer
|
||||
try
|
||||
{
|
||||
var consumerKey = $"{groupId}_{string.Join("_", topics)}_{typeof(TKey).Name}_{typeof(TValue).Name}";
|
||||
if (_consumerStore.TryGetValue(consumerKey, out var entry))
|
||||
if (_consumerStore.TryRemove(consumerKey, out var entry))
|
||||
{
|
||||
entry.CTS.Cancel();
|
||||
(entry.Consumer as IDisposable)?.Dispose();
|
||||
entry.CTS.Dispose();
|
||||
// 从字典中移除
|
||||
_consumerStore.TryRemove(consumerKey, out entry);
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user