Java: несколько потоков против сокетов

Question

Я написал простое приложение на Java, в котором есть два узла, каждый с ServerSocket, открытым для порта, который прослушивает входящие соединения. Узлы запускают два потока каждый, посылая 1000 сообщений другому узлу через постоянный сокет TCP, созданный при отправке первого сообщения. Однако узлы не получают все 1000 сообщений. Один может получить 850, а другой - только 650. Это число имеет тенденцию оставаться постоянным в течение нескольких прогонов.

Код отправки:

public void SendMsg(String dest, Message myMsg) {
    Socket sendsock = null;
    PrintWriter printwr = null;
    try {
        if(printwr == null) {
            sendsock = new Socket(dest, Main.rcvport);
            printwr = new PrintWriter(sendsock.getOutputStream(), true);
        }
        String msgtosend = myMsg.msgtype.toString() + "=" + Main.myaddy + "=" + myMsg.content + "\n";
        printwr.print(msgtosend);
    } catch (UnknownHostException ex) {
        System.out.println(ex);
        //DO: Terminate or restart
    } catch (IOException ex) {
        System.out.println(ex);
        //DO: Terminate or restart
    }
}

Производительность, кажется, улучшается, если я использую buffwr = новый BufferedWriter (printwr) также используйте buffwr.write (...) вместо printwr.print (...) , хотя это не кажется полным решением для потери данных , Нет никаких исключений, чтобы показать, что пакеты не были доставлены, поэтому, согласно отправителю, все они были успешно отправлены.

На принимающей стороне принятое соединение обрабатывается следующим образом:

BufferedReader inbuff = new BufferedReader(new InputStreamReader(incoming.getInputStream()));

        while(running) {
            String rcvedln = inbuff.readLine();
            if(rcvedln != null) {
                count++;
                System.out.println(count);
            }
        }

Есть ли проблема с использованием читателей и писателей, которая могла быть причиной проблемы? Спасибо.

Answer 1

SendMsg() создает новый сокет при каждом вызове, поэтому вы не используете постоянное TCP-соединение. Метод также не закрывает сокет, поэтому у вас есть много открытых коллекций. Возможно, вы достигли предела числа соединений, которые может установить процесс (сокеты могут не закрываться при сборке мусора).

Наконец, как указывало kd304, Javadoc для PrintWriter утверждает это о параметре autoFlush конструктора PrintWriter: «если true, методы println, printf или format сбрасывают буфер вывода». Ваш код не вызывал метод, который сделал сброс.

Попробуйте это:

public class MessageSender implements Closeable {
  private final Socket socket;
  private final PrintWriter writer;

  public MessageSender(String dest, int port) {
    socket = new Socket(dest, port);
    writer = new PrintWriter(socket.getOutputStream(), true);
  }

  public void sendMessage(Message message) {
    try {
        writer.println(message.toString());
    } catch (UnknownHostException ex) {
        System.out.println(ex);
        //DO: Terminate or restart
    } catch (IOException ex) {
        System.out.println(ex);
        //DO: Terminate or restart
    }
}

@Override
public void close() throws IOException {
  writer.close();
  socket.close();
}

Примечание. Я изменил код так, чтобы sendMessage() вызывал Message.toString() для получения отформатированного сообщения. sendMessage() кажется неправильным ссылаться на поля в Message для форматирования сообщения. Вместо использования toString() вы можете создать метод в Message специально для этой цели.

Вот код на стороне сервера:

public class Server implements Runnable {
  private final ServerSocket serverSocket;
  private final ExecutorService executor;
  private volatile boolean running = true;

  public Server(int port, ExecutorService executor) throws IOException {
    serverSocket = new ServerSocket(port);
    this.executor = executor;
  }

  @Override
  public void run() throws IOExeption {
    while (running) {
      Socket socket = serverSocket.accept();
      executor.execute(new ConnectionHandler(socket));
    }
  }

  public boolean stop(long timeout, TimeUnit unit) {
    running = false;
    executor.shutdown();
    return executor.awaitTermination(timeout, unit);
  }
}

Вы можете использовать Executors, чтобы создать ExecutorService для запуска задач. Обратите внимание, что ConnectionHandler должен закрыть заданный сокет.

Answer 2

Вы закрываете PrintWriter для очистки потока?

} finally {
    printwr.close();
    sendsock.close();
}

Answer 3

Если я помещу Thread.sleep (2) в цикл for, где вызывается функция SendMsg, больше сообщений будет получено правильно, но это не всегда 1000. Возможно ли, что системные ресурсы перегружены двумя потоками работает ли цикл постоянно?

Answer 4

Ах, прости. Я случайно удалил комментирование из кода. Это на самом деле так:

public void SendMsg(String dest, Message myMsg) {
Socket sendsock = null;
try {
    if(printwr == null) {
        sendsock = new Socket(dest, Main.rcvport);
        printwr = new PrintWriter(sendsock.getOutputStream(), true);
    }
    String msgtosend = myMsg.msgtype.toString() + "=" + Main.myaddy + "=" + myMsg.content + "\n";
    printwr.print(msgtosend);
} catch (UnknownHostException ex) {
    System.out.println(ex);
    //DO: Terminate or restart
} catch (IOException ex) {
    System.out.println(ex);
    //DO: Terminate or restart
}

}

printrw объявляется и сохраняется вне функции, поэтому после его настройки нет необходимости в sendsock или в повторной инициализации printrw. В настоящем приложении я сохраняю PrintWriter для каждого соединения в HashMap и извлекаю его в начале функции SendMsg (...).

Поскольку соединения постоянны, каждый раз, когда один из них принимается, новый поток - это ланч, который запускает цикл while, чтобы непрерывно проверять его на наличие данных. Эти потоки и соединения закрываются только после завершения работы приложения. В дополнение к моему предыдущему вопросу, есть ли более эффективный способ сделать это?

Ранее я реализовал этот код без "\ n" и использовал вместо него println (...), и у меня все еще оставалась проблема с получением некоторых сообщений, поэтому я не уверен, что является причиной проблемы , Сообщения отправляются так:

public class SendPortal2 implements Runnable {
String dest = null;

SendPortal2 (String dest) {
    this.dest = dest;
}

public void run() {
        for(int i=1; i<1000; i+=2) {
            Message myMsg = new Message("Message", Main.myaddy + " " + String.valueOf(i));
            Main.myCommMgr.SendMsg(dest, myMsg);
        }
}

}

Работают два таких потока. Когда я только что снова запустил код, одна сторона получила 999 пакетов, а другая - только 500, что заставило меня поверить, что иногда данные из целого потока могут быть заблокированы. Это вероятно?

Спасибо за ответы!