【问题标题】:How to handle and perform actions fast on websocket data in c#?如何在 c# 中对 websocket 数据进行快速处理和执行操作?
【发布时间】:2019-03-07 19:01:56
【问题描述】:

我正在使用 websocket 连接到第三方数据馈送提供商的服务器。 对于 websocket 连接,我的代码是:

this.websocket = new WebSocket("wss://socket.polygon.io/stocks", sslProtocols: SslProtocols.Tls12 | SslProtocols.Tls11 | SslProtocols.Tls);

因此,当建立连接时,我们每分钟都会收到近 70,000 到 1,00,000 条记录。因此,在那之后,我们将这些响应分叉并将其存储在单独的文件中。就像我们收到 AAPL 的数据一样,我们将这些数据存储到 AAPL 的文件中。与 FB、MSFT、IBM、QQQ 等相同。我们总共需要处理 10,000 个文件,并根据这些文件存储实时记录。

public static string tempFile = @"D:\TempFileForLiveMarket\tempFileStoreLiveSymbols.txt";
public static System.IO.StreamWriter w;
private void websocket_MessageReceived(object sender, MessageReceivedEventArgs e)
{
  using (w = System.IO.File.AppendText(tempFile))
  {
     Log(e.Message, w);
  }
  using (System.IO.StreamReader r = System.IO.File.OpenText(tempFile))
  {
     DumpLog(r);
  }
}

public static void Log(string responseMessage, System.IO.TextWriter w)
{
     w.WriteLine(responseMessage);
}

public static void DumpLog(System.IO.StreamReader r)
{
  string line;
  while ((line = r.ReadLine()) != null)
  {
     WriteRecord(line);
  }
}

public static void WriteRecord(string data)
{
   List<LiveData> ld = JsonConvert.DeserializeObject<List<LiveData>>(data);
   var filterData = ld.Where(x => symbolList.Contains(x.sym));
   List<string> fileLines = new List<string>();
   foreach (var item in filterData)
   {
      var fileName = @"D:\SymbolsData\"+item.sym+ "_day_Aggregate.txt";
      fileLines = File.ReadAllLines(fileName).AsParallel().Skip(1).ToList();
      if (fileLines.Count > 1)
      {
         var lastLine = fileLines.Last();
         if (!lastLine.Contains(item.sym))
         {
               fileLines.RemoveAt(fileLines.Count - 1);
         }
      }
      fileLines.Add(item.sym + "," + item.s + "," + item.p + "-----");
      System.IO.File.WriteAllLines(fileName, fileLines);
   }
}

因此,当 websocket 连接建立并使用我们的 10,000 个文件对实时市场数据执行操作时,它会变得更慢,并且 websocket 连接会在几分钟后关闭并传递如下消息:

Websocket Error
Received an unexpected EOF or 0 bytes from the transport stream.
Connection Closed...

我正在执行整个过程,因为在下一阶段我需要对每个交易品种的实时价格进行技术分析。那么我该如何处理这种情况呢?我怎样才能使处理速度比这个处理速度更快?以及如何停止连接关闭?

编辑后

我用字符串生成器替换流写入器和临时文件,如下所示,

public static StringBuilder sb = new StringBuilder();
public static System.IO.StringWriter sw;
private void websocket_MessageReceived(object sender, MessageReceivedEventArgs e)
{
      sw = new System.IO.StringWriter(sb);
      sw.WriteLine(e.Message);
      Reader();
}

public static void Reader()
{
     System.IO.StringReader _sr = new System.IO.StringReader(sb.ToString());
     while (_sr.Peek() > -1)
     {
            WriteRecord(sb.ToString());
     }
     sb.Remove(0, sb.Length);
}

public static void WriteRecord(string data)
{
     List<LiveData> ld = JsonConvert.DeserializeObject<List<LiveData>>(data);
     foreach (var item in filterData)
     {
           var fileName = @"D:\SymbolsData\"+item.sym+ "_day_Aggregate.txt";
           fileLines = File.ReadAllLines(fileName).AsParallel().Skip(1).ToList();
           fileLines.RemoveAt(fileLines.Count - 1);
           fileLines.Add(item.sym + "," + item.s + "," + item.p)
           System.IO.File.WriteAllLines(fileName, fileLines);
      }
}

【问题讨论】:

  • 无论你做什么,写入文件都会很慢。也许缓冲消息,然后稍后再写入 - 这样你可以避免文件 io
  • @PatrickHollweck,是的,也可以将实时数据存储到临时文件中,并为每 1 分钟设置一次计时器,这将每 1 分钟从临时文件中收集数据,然后对其进行处理。但是,如果可能使用实时数据,那么使用实时价格会更好。在我们的选项中,它将处理 1 分钟的延迟价格数据。所以我首先推荐这个选项。
  • 将数据保存到数据库中会比保存大文件更快。像 SQL Server 这样的数据库旨在处理实时数据存储。打开文件时会占用大量内存。当您附加到文件时,它需要永远打开并移动到 Windows 中的文件末尾。随着文件大小变大,移动到结束的时间也会变大。
  • @jdweng,是的,这更好,但我正在处理每个品种的日数据、分钟数据、报价数据和交易数据,每个部分处理近 25,000 条记录。因此,对于每个符号,都会有 1 条 lac 记录。我将拥有总共 10,669 个文件。所以可能这种类型的非常大的数据很难在任何数据库上处理,所以我尝试使用内存映射文件。
  • 内存映射永远不会起作用。数据库旨在处理大量数据。看起来您只有 2.5 亿条记录。如果将数据放入一个大文件中,读取文件需要多长时间?大内存映射最终会导致数据在驱动器上非常缓慢的交换空间中结束。

标签: c# linq sockets websocket filestream


【解决方案1】:

看起来您将每条消息附加tempFile,但随后您处理整个tempFile。这意味着您不断地重新处理数据加上新记录,所以是的:它会逐渐花费越来越长的时间,直到它需要很长时间,另一个end 厌倦了等待,把你打断了。我的建议:不要那样做。

在实际处理每条记录的过程中,您还可以很多做一些更有效的事情,但这与不断重新处理所有事情的开销相比是无关紧要的.

【讨论】:

  • 感谢您的宝贵建议。如果我们使用 StringBuilder 那么它会影响进程而不是 StreamWrite 和 tempfile 吗?
  • @Ankit 那...不是重点;关键是您当前正在创建并重新处理一个越来越大的文件;现在...是的,显然如果您不接触文件系统,事情会更快,但大概您想要文件系统上的数据。问题是你一直在重新处理一切。所以...不要那样做?
  • 我们的 WriteRecord 方法无法将每个响应写入特定的符号文件。我们每分钟收到 80,000 条记录,我们有 10,000 个符号文件。所以响应非常快,将所有记录写入文件非常慢。所以临时文件的主要目的是如果在符号文件中写入记录变慢,那么我们不能丢失它。我们可以从 tempfile 中收集它。但是,如果我们可以在符号文件中获得非常高的写入响应速度,那么我们也不需要 tempfile 和 stringbuilder。那么有什么方法可以让 WriteRecord 方法的处理速度非常快?
猜你喜欢
  • 1970-01-01
  • 2011-01-10
  • 1970-01-01
  • 1970-01-01
  • 2020-09-30
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
相关资源
最近更新 更多