2

我正在寻找从线路中读取数据并将其发送到函数以解析内容。由于消息 (xml) 的大小可能不同,我可以阅读整条消息、多条消息或消息部分。

我正在尝试使用当我从网络上读取数据并使用消费者线程将数据拉出以进行解析时BlockingCollection调用的地方来实现代码。这些示例看起来很简单,但它们似乎只工作一次,然后退出。我希望消费者在消息进入时不断解析。请参阅下面的代码了解我目前在做什么。TryAddBlockingCollection

消息处理:

    private static BlockingCollection<byte[]> queue = new BlockingCollection<byte[]>();

    public XmlHandler()
    {
        CancellationTokenSource cts = new CancellationTokenSource();
        Task.Factory.StartNew(() =>
      {
          if (Console.ReadKey().KeyChar == 'c')
              cts.Cancel();
      });


        Task.Factory.StartNew(() => ParserWorker(queue, cts.Token));

    }
    //run producer
    public void AddData(byte[] data, int bytesRead)
    {
        bool success = false;
        try
        {
            success = queue.TryAdd(data);
        }
        catch (OperationCanceledException)
        {
            Console.WriteLine("Add loop canceled.");
            queue.CompleteAdding();
        }

        if (success)
        {
            Console.WriteLine(" Add:{0}", data);
        }
        else
        {
            Console.Write(" AddBlocked");
        }

        System.Console.WriteLine("queue count = " + queue.Count);
    }

    private static void ParserWorker(BlockingCollection<byte[]> bc, CancellationToken ct)
    {
        ASCIIEncoding encoder = new ASCIIEncoding();
        String xmlString;

        while (!bc.IsCompleted)
        {
            byte[] nextItem;
            try
            {

                if (!bc.TryTake(out nextItem, 0, ct))
                {
                    Console.WriteLine(" Take Blocked");
                }
                else
                {
                    xmlString = encoder.GetString(nextItem, 0, nextItem.Length);
                    System.Console.WriteLine(xmlString);
                }

            }

            catch (OperationCanceledException)
            {
                Console.WriteLine("Taking canceled.");
                break;
            }
        }       
    }

读取电线(这在一个线程中运行):

private void HandleClientComm(object client)
{
      TcpClient tcpClient = (TcpClient)client;
      NetworkStream clientStream = tcpClient.GetStream();

      byte[] message = new byte[8192];
      int bytesRead;

      while (true)
      {
          bytesRead = 0;

          try
          {

              bytesRead = clientStream.Read(message, 0, 4096);
              byte[] temp = new byte[bytesRead];
              Array.Copy(message, temp, bytesRead);

              /*CODE WILL HANG HERE...*/
              ASCIIEncoding encoder = new ASCIIEncoding();
              String xmlString = encoder.GetString(message, 0, message.Length);
              System.Console.WriteLine(xmlString);
              /*DOES NOT GO BEYOND LINE ABOVE */

              handler.AddData(message, bytesRead);  //xmlhandler
           }
           catch (Exception e)
           {
               System.Console.WriteLine(e.ToString());
               break;
           }

           if (bytesRead == 0)
           {
               break;
           }
      }
}

那么谁能告诉我我在这里做错了什么?

4

0 回答 0