- Worker: 风险/预付金计算由连续空转改为30秒一轮(可被取消令牌立即唤醒) - ClientPosiTask: HasNewFlow 改为与上次计算时间比较(原误传当前时间致增量判断失效),轮询间隔2秒 - 30秒兜底全量重算语义保留不变
74 lines
2.9 KiB
C#
74 lines
2.9 KiB
C#
using System;
|
|
using System.Collections.Generic;
|
|
using System.Linq;
|
|
using System.Text;
|
|
using System.Threading.Tasks;
|
|
using YLErp.Abstract;
|
|
using YLErp.BLL.Eod;
|
|
using YLErp.Cache;
|
|
using YLErp;
|
|
using Microsoft.Extensions.Caching.Memory;
|
|
|
|
namespace RealTimeCalcPositionService
|
|
{
|
|
public class ClientPosiTask : IHostedService, IDisposable
|
|
{
|
|
private readonly IYcLogger _logger;
|
|
private IYLCache _yLCache;
|
|
private readonly CancellationTokenSource _cts = new CancellationTokenSource();
|
|
private IKafkaProduce kafkaProduceHelper;
|
|
IMemoryCache memoryCache;
|
|
public ClientPosiTask(IYLCache yLCache, IKafkaProduce kafkaProduce)
|
|
{
|
|
_logger = LogFactory.GetLogger("Worker");
|
|
_yLCache = yLCache;
|
|
kafkaProduceHelper = kafkaProduce;
|
|
memoryCache = new MemoryCache(new MemoryCacheOptions());
|
|
RealtimePnlCalc.InitCache(_yLCache);
|
|
RealtimePnlCalc.InitKafka(kafkaProduceHelper);
|
|
}
|
|
|
|
public Task StartAsync(CancellationToken cancellationToken)
|
|
{
|
|
_logger.Info($"实时持仓服务开始:{DateTimeOffset.Now}");
|
|
Task.Run(() => ExecuteTask(_cts.Token), _cts.Token);
|
|
return Task.CompletedTask;
|
|
}
|
|
private void ExecuteTask(CancellationToken stoppingToken)
|
|
{
|
|
while (!stoppingToken.IsCancellationRequested)
|
|
{
|
|
string calctime = memoryCache.Get<string>("calctime");
|
|
var currentTime = DateTime.Now;
|
|
//以上次计算时间为基准判断是否有新流水,避免无变化时空转重算
|
|
var hasNew = DateTime.TryParseExact(calctime, "yyyy-MM-dd HH:mm:ss.fff",
|
|
System.Globalization.CultureInfo.InvariantCulture, System.Globalization.DateTimeStyles.None, out var lastCalcTime)
|
|
&& RealtimePnlCalc.HasNewFlow(lastCalcTime);
|
|
if (string.IsNullOrEmpty(calctime)|| hasNew)
|
|
{
|
|
memoryCache.Set("calctime", currentTime.ToString("yyyy-MM-dd HH:mm:ss.fff"), TimeSpan.FromSeconds(30));
|
|
try
|
|
{
|
|
RealtimePnlCalc.RealtimeSwapPosition(new OptUserInfo(0, "互换实时持仓服务", OptUserFrom.Service));
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.Error(ex, "实时持仓服务异常:" + ex.Message);
|
|
}
|
|
}
|
|
stoppingToken.WaitHandle.WaitOne(TimeSpan.FromSeconds(2));
|
|
}
|
|
}
|
|
public async Task StopAsync(CancellationToken cancellationToken)
|
|
{
|
|
_logger.Info($"实时持仓服务停止: {DateTimeOffset.Now}");
|
|
_cts.Cancel();
|
|
await Task.CompletedTask;
|
|
}
|
|
public void Dispose()
|
|
{
|
|
_cts.Dispose();
|
|
}
|
|
}
|
|
}
|