问题描述
SubscribetoKlineUpdatesAsync 是一个网络套接字流事件处理程序，每秒从 Binance 接收数据。我在里面有一个长时间运行的操作 10 秒。它被分成一个任务，因为我不希望它阻止任何传入的价格数据。
问题是它一次生成数千个任务，我希望它每次只生成一个符号一个任务。该符号由 data.Symbol 标识。在代码下方，您将找到日志。第二个日志是我想要的样子。
我想使用像 TPL Dataflow 或 Rx.NET 这样的东西。我只是不知道该怎么做。我知道 lock 的方式，BlockingCollection 和 SemaphoreSlim。
下面有一个日志，记录了当前情况和我想要完成的任务。
代码
using System;
using System.Threading;
using System.Threading.Tasks;
using Binance.Net;
using Binance.Net.Enums;
using Binance.Net.Interfaces;
using Binance.Net.Objects.Spot;
using CryptoExchange.Net.Authentication;
using CryptoExchange.Net.Logging;
namespace SubscribetoCandlesEventFixTest
{
public class BinanceSocketHandler
{
private readonly IBinanceClient _client;
private readonly IBinanceSocketClient _socketClient;
public BinanceSocketHandler()
{
_client = new BinanceClient(new BinanceClientOptions
{
ApiCredentials = new ApiCredentials("not required","not required"),AutoTimestamp = true,AutoTimestampRecalculationInterval = TimeSpan.FromMinutes(30),#if DEBUG
LogVerbosity = LogVerbosity.Debug
#endif
});
_socketClient = new BinanceSocketClient(new BinanceSocketClientOptions
{
ApiCredentials = new ApiCredentials("not required",AutoReconnect = true,ReconnectInterval = TimeSpan.FromSeconds(15),#if DEBUG
LogVerbosity = LogVerbosity.Debug
#endif
});
}
public Task StartAsync(CancellationToken cancellationToken)
{
var symbols = new[] { "TRXUSDT","BTCUSDT" };
var interval = KlineInterval.OneMinute;
return _socketClient.Spot.SubscribetoKlineUpdatesAsync(symbols,interval,data =>
{
if (data.Data.Final)
{
Console.WriteLine($"[{DateTime.UtcNow}] [{data.Symbol}] New final candle | Timestamp: {data.Data.OpenTime} | Price: {data.Data.Close}");
}
else
{
Console.WriteLine($"[{DateTime.UtcNow}] [{data.Symbol}] Candle update | Timestamp: {data.Data.OpenTime} | Price: {data.Data.Close}");
Task.Run(async () =>
{
Console.WriteLine("Operation taking 10 seconds to execute...");
await Task.Delay(10000,cancellationToken).ConfigureAwait(false);
},cancellationToken);
}
});
}
}
class Program
{
static async Task Main(string[] args)
{
var test = new BinanceSocketHandler();
await test.StartAsync(new CancellationToken()).ConfigureAwait(false);
Console.ReadLine();
}
}
}
日志
它产生了数千个任务。每个符号我只想要一个任务。
[3/7/2021 12:24:53 AM] [TRXUSDT] Candle update | Timestamp: 3/7/2021 12:24:00 AM | Price: 0.05049000
Operation taking 10 seconds to execute...
[3/7/2021 12:24:53 AM] [BTCUSDT] Candle update | Timestamp: 3/7/2021 12:24:00 AM | Price: 49240.17000000
Operation taking 10 seconds to execute...
[3/7/2021 12:24:55 AM] [BTCUSDT] Candle update | Timestamp: 3/7/2021 12:24:00 AM | Price: 49245.66000000
Operation taking 10 seconds to execute...
[3/7/2021 12:24:57 AM] [BTCUSDT] Candle update | Timestamp: 3/7/2021 12:24:00 AM | Price: 49251.02000000
Operation taking 10 seconds to execute...
[3/7/2021 12:24:59 AM] [BTCUSDT] Candle update | Timestamp: 3/7/2021 12:24:00 AM | Price: 49254.44000000
Operation taking 10 seconds to execute...
[3/7/2021 12:25:00 AM] [TRXUSDT] Candle update | Timestamp: 3/7/2021 12:24:00 AM | Price: 0.05049000
Operation taking 10 seconds to execute...
[3/7/2021 12:25:01 AM] [BTCUSDT] New final candle | Timestamp: 3/7/2021 12:24:00 AM | Price: 49257.99000000
[3/7/2021 12:25:01 AM] [TRXUSDT] New final candle | Timestamp: 3/7/2021 12:24:00 AM | Price: 0.05049000
[3/7/2021 12:25:03 AM] [BTCUSDT] Candle update | Timestamp: 3/7/2021 12:25:00 AM | Price: 49264.26000000
Operation taking 10 seconds to execute...
[3/7/2021 12:25:04 AM] [TRXUSDT] Candle update | Timestamp: 3/7/2021 12:25:00 AM | Price: 0.05049000
Operation taking 10 seconds to execute...
[3/7/2021 12:25:05 AM] [BTCUSDT] Candle update | Timestamp: 3/7/2021 12:25:00 AM | Price: 49266.60000000
Operation taking 10 seconds to execute...
[3/7/2021 12:25:07 AM] [BTCUSDT] Candle update | Timestamp: 3/7/2021 12:25:00 AM | Price: 49251.02000000
Operation taking 10 seconds to execute...
[3/7/2021 12:25:07 AM] [TRXUSDT] Candle update | Timestamp: 3/7/2021 12:25:00 AM | Price: 0.05050000
Operation taking 10 seconds to execute...
我想要的示例日志
[3/7/2021 12:24:53 AM] [TRXUSDT] Candle update | Timestamp: 3/7/2021 12:24:00 AM | Price: 0.05049000
Operation taking 10 seconds to execute...
[3/7/2021 12:24:53 AM] [BTCUSDT] Candle update | Timestamp: 3/7/2021 12:24:00 AM | Price: 49240.17000000
Operation taking 10 seconds to execute...
[3/7/2021 12:24:55 AM] [BTCUSDT] Candle update | Timestamp: 3/7/2021 12:24:00 AM | Price: 49245.66000000
[3/7/2021 12:24:57 AM] [BTCUSDT] Candle update | Timestamp: 3/7/2021 12:24:00 AM | Price: 49251.02000000
[3/7/2021 12:24:59 AM] [BTCUSDT] Candle update | Timestamp: 3/7/2021 12:24:00 AM | Price: 49254.44000000
[3/7/2021 12:25:00 AM] [TRXUSDT] Candle update | Timestamp: 3/7/2021 12:24:00 AM | Price: 0.05049000
[3/7/2021 12:25:01 AM] [BTCUSDT] New final candle | Timestamp: 3/7/2021 12:24:00 AM | Price: 49257.99000000
[3/7/2021 12:25:01 AM] [TRXUSDT] New final candle | Timestamp: 3/7/2021 12:24:00 AM | Price: 0.05049000
[3/7/2021 12:25:03 AM] [BTCUSDT] Candle update | Timestamp: 3/7/2021 12:25:00 AM | Price: 49264.26000000
Operation taking 10 seconds to execute...
[3/7/2021 12:25:04 AM] [TRXUSDT] Candle update | Timestamp: 3/7/2021 12:25:00 AM | Price: 0.05049000
Operation taking 10 seconds to execute...
[3/7/2021 12:25:05 AM] [BTCUSDT] Candle update | Timestamp: 3/7/2021 12:25:00 AM | Price: 49266.60000000
[3/7/2021 12:25:07 AM] [BTCUSDT] Candle update | Timestamp: 3/7/2021 12:25:00 AM | Price: 49251.02000000
[3/7/2021 12:25:07 AM] [TRXUSDT] Candle update | Timestamp: 3/7/2021 12:25:00 AM | Price: 0.05050000
解决方法
暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！
如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。
小编邮箱:dio#foxmail.com (将#修改为@）