Async Отправить нескольким клиентам и правильно использовать FromAsync?

Question

У меня 500 подключенных клиентов на сервере. Сервер отправляет данные всем клиентам за каждую секунду.

Когда сервер отправляет данные клиентам, результат:

-----------Result-----------
Client1 received : 102 datas
Client2 received : 109 datas
Client3 received : 105 datas
Client4 received : 108 datas
Client5 received : 108 datas
ClientN received : 107 datas

Но в конце должно быть 110 данных.

Я использую следующий код для отправки данных клиентам

public static class AsyncSend
{
    public static Task<int> SendAsync(this Socket socket, byte[] buffer, int offset, int count)
    {
        return new Task<int>(() =>
        {
            return Task.Factory.FromAsync<int>(
                           socket.BeginSend(buffer, offset, count, SocketFlags.None, null, socket),
                     socket.EndSend).Result;
        });
    }
}

class ServerTest
{
  private Thread sendThread;
  private Dictionary<String, Client> connectedClients;

  public ServerTest()
  {
    connectedClients = new Dictionary<String, Client>();
  }

  public void receive()
  {
    //here server receives data from another server
    //When server receives new data then fires the newDataReceived(String data) method.

    newDataReceived(data);
  }

  public void newDataReceived(String data)
  {
    sendThread = new Thread(new ParameterizedThreadStart(sendToClients));
    sendThread.Start(data);
  }

  private void sendToClients(object dataObj)
  {
    try
    {
      int totalConnectedClients = connectedClients.Count;

      if (totalConnectedClients > 0)
      {
        lock (connectedClients)
        {
          byte[] buffer = Encoding.UTF8.GetBytes((string)dataObj);
          byte[] cBuffer = new SCompressor().Compress(buffer);

          foreach (Client client in connectedClients.Values)
          {
            Task<int> task = AsyncSend.SendAsync(client.Socket, cBuffer, 0, cBuffer.Length);
            task.Start();
          }
        }
      }
    }
    catch (Exception ex)
    {
      //ERROR.
    }
  }

Есть ли другой способ или как правильно и эффективно отправлять данные нескольким клиентам?

Пожалуйста, покажи мне примеры, если можешь.

Спасибо за совет.

Answer 1

У меня есть несколько очков для вас:

Вы можете повысить эффективность, избавившись от одной лишней задачи за отправку:

public static Task<int> SendAsync(this Socket socket, byte[] buffer, int offset, int count)
{
        return Task.Factory.FromAsync<int>(
                       socket.BeginSend(buffer, offset, count, SocketFlags.None, null, socket),
                 socket.EndSend);
//Don't explicitly start the return value of this method. The task will already be started.
}

Вы используете асинхронную отправку, насколько ямогу сказать (кроме ненужной задачи).

Используете ли вы UDP?Если да, это объясняет, почему некоторые данные отсутствуют.

Есть ли обнаруженная ошибка?Если да, цикл будет прерван, и не все клиенты получат свои данные.

У вас есть условие гонки:

int totalConnectedClients = connectedClients.Count;

Это должно происходить внутри блокировки.Вы не можете безопасно считать счетчик, в то время как другие темы могут обновить этот словарь.

Как вы получаете данные?Возможно ошибка находится на стороне приема.Пожалуйста, отправьте код.

Редактировать:

Что делать, если запущены два одновременных приема?Это может привести к ошибке: клиенты могут получать сначала самые новые данные, а затем более старые данные.Я рекомендую вам не начинать новый поток каждый раз, когда вы получаете данные.Вместо этого, постоянно работающий поток, который извлекает новые данные из коллекции BlockingCollection с помощью метода GetConsumerEnumerable.Ваш метод newDataReceived будет просто помещен в эту коллекцию BlockingCollection.Вы можете найти отличное введение здесь: http://blogs.msdn.com/b/csharpfaq/archive/2010/08/12/blocking-collection-and-the-producer-consumer-problem.aspx

Answer 2

Я абсолютно не представляю, что означает «102 данных» или что вы на самом деле спрашиваете, но ваш SendAsync() метод может быть значительно улучшен, например:

public static Task<int> SendAsync(this Socket socket, byte[] buffer, int offset, int count)
{
    return Task.Factory.FromAsync<int>(
        socket.BeginSend(buffer, offset, count, SocketFlags.None, null, socket),
        socket.EndSend);
}

Таким образом, вы не блокируете ни один поток, и поэтому вы должны повысить эффективность.

Одно из различий между измененным кодом и вашим кодом заключается в том, что Task уже запущен, поэтому вам не следует вызывать Start() для него.

Answer 3

Одна вещь, которая не была сделана в этом потоке, - то, что TCP является потоковым протоколом.Это означает, что если сервер отправляет 110 байтов, клиент может получить что-нибудь между 1-110 байтами за один вызов чтения.Если клиент знает, что ожидается 110 байт, ему необходимо повторно вызывать Read, пока не поступят все 110 байтов.

Answer 4

Добавляет ли класс SCompressor префикс длины или уникальный конечный маркер к данным, которые он возвращает?

Без этого клиентский код не сможет узнать, сколько байтов получить до распаковки.

Это единственная потенциальная ошибка, которая, как я вижу, может повлиять на объем данных , получаемых клиентами, при условии, что это то, что показывает диагностика результатов клиента.

Если это так легко исправить - просто добавьте длину сжатых данных в начало буфера, который отправляет сервер:

        var sendBuffer = new byte[cBuffer.Length + 4];
        BitConverter.GetBytes(cBuffer.Length).CopyTo(sendBuffer, 0);
        cBuffer.CopyTo(sendBuffer, 4);

Затем измените клиент на чтение 4 байтов, декодируйте длину как int, продолжайте чтение до тех пор, пока не получите длина байтов, распакуйте.