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 
    } 
} 

Performance улучшить, если я использую 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

Вы закрываете PrintWriter, чтобы очистить поток?

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

Answer 2

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

Наконец, как kd304 указал, Javadoc для PrintWriter состояний этого о параметре PrintWriter конструктора autoFlush: «если это правда, то Println, Printf или методы форматирования смоет выходной буфер». Ваш код не вызывал метод, который делал флеш.

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

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 3

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

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, что заставило меня поверить, что данные из всего потока могут быть заблокированы. Это возможно?

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

Answer 4

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