57 lines
2.2 KiB
C#
57 lines
2.2 KiB
C#
using Microsoft.Extensions.DependencyInjection;
|
|
using NCrontab;
|
|
using YLErp.Modules.AppModule;
|
|
using YLErp.Modules.SwapModule;
|
|
|
|
namespace YLErp.Web.App
|
|
{
|
|
public class SendKafkaTask : IHostedService, IDisposable
|
|
{
|
|
private readonly IYcLogger _logger;
|
|
private readonly CancellationTokenSource _cts = new CancellationTokenSource();
|
|
private readonly string _cronExpression = "";
|
|
private CrontabSchedule _schedule; // cron 表达式解析后的时间表
|
|
public SendKafkaTask(IConfiguration configuration)
|
|
{
|
|
_logger = LogFactory.GetLogger("SendKafkaTask");
|
|
_cronExpression = configuration.GetValue<string>("SendKafakaCron");
|
|
_schedule = CrontabSchedule.Parse(_cronExpression);
|
|
}
|
|
public Task StartAsync(CancellationToken cancellationToken)
|
|
{
|
|
_logger.Info("SendKafkaTask task is starting.");
|
|
Task.Run(() => ExecuteTask(_cts.Token), _cts.Token);
|
|
return Task.CompletedTask;
|
|
}
|
|
public async Task StopAsync(CancellationToken cancellationToken)
|
|
{
|
|
_logger.Info("SendKafkaTask task is stopping.");
|
|
_cts.Cancel();
|
|
await Task.CompletedTask;
|
|
}
|
|
public void Dispose()
|
|
{
|
|
_cts.Dispose();
|
|
}
|
|
private async Task ExecuteTask(CancellationToken cancellationToken)
|
|
{
|
|
while (!cancellationToken.IsCancellationRequested)
|
|
{
|
|
var now = DateTime.Now;
|
|
var nextOccurrence = _schedule.GetNextOccurrence(now);
|
|
var delay = nextOccurrence - now;
|
|
if (delay.TotalMilliseconds > 0)
|
|
{
|
|
_logger.Info($"Waiting SendKafkaTask {delay.TotalMilliseconds} ms until {nextOccurrence}");
|
|
await Task.Delay(delay, cancellationToken);
|
|
}
|
|
// 执行任务
|
|
await Task.Run(async () => {
|
|
new SwapRateService(OptUserInfo.SystemUser).SendAllSwapRateToKafka();
|
|
new EtradingRuleService(OptUserInfo.SystemUser).SendAllEtradingRuleToKafka();
|
|
}, cancellationToken);
|
|
}
|
|
}
|
|
}
|
|
}
|