Java：多线程与套接字

Question

我用 Java 编写了一个简单的应用程序，其中有两个节点，每个节点都有一个 ServerSocket，它打开一个端口，监听传入的连接。每个节点运行两个线程，通过在发送第一条消息时创建的持久 TCP 套接字向另一个节点发送 1000 条消息。但是，节点不会收到全部 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 = new 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 所指出的，PrintWriter 的 Javadoc 说明了 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 中的字段来格式化消息。您可以在Message 中专门为此目的创建一个方法，而不是使用toString()。

这是服务器端代码：

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

啊，对不起。我不小心从代码中删除了注释。其实是这样的：

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 是在函数外部声明和存储的，所以一旦设置好，就不需要 sendock 或重新初始化 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

如果我在调用 SendMsg 函数的 for 循环中放置一个 Thread.sleep(2) ，则可以正确接收更多消息，但并不总是 1000 条。是否有可能系统的资源被两个线程占用连续运行while循环？