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