java - 从不同线程收集数据

Question

我已经实现了一个基本的 TCP 客户端和服务器。客户端从输入流中发送命令。服务器处理收到的消息并回复客户端。

我想以下列方式测试我的客户端-服务器解决方案:

创建多个客户端线程；

每个客户端线程将从文件中读取命令并将它们发送到服务器；

在服务器回复后，每个客户端必须将回复收集到一个列表中；

我想寻求有关如何从每个客户端线程收集服务器对列表的回复的任何代码审查和任何建议(我将不胜感激任何示例)。目前，我无法收集 List<String> messages 中所有客户端线程的所有回复。

类(class) Client产生固定数量的 ClientTask

public class Client {
private static Logger logger = Logger.getLogger(Client.class);
private File configs;
private InputStream inputStream;
private int port;
private String ip;


public Client(File file, InputStream inputStream) {
    this.configs = file;
    this.inputStream = inputStream;
    Map<String, Object> configs = ConfigLoader.loadXMLConfigsFromFile(file);
    this.port = (Integer) configs.get("port");
    this.ip = (String) configs.get("ip");
}

//
public void start(int numberOfThreads, List<String> messages) throws InterruptedException {
    for (int i = 0; i < numberOfThreads; i++) {
        List<String> collectedServerReply = Collections.synchronizedList(new ArrayList<>());
        try {
            Thread clientThread = new Thread(
                    new ClientTask(port, ip)
                            .setInputStream(new FileInputStream("commands.txt"))
                            .setReplyListener(message -> {
                                //messages.add(message.receive().getMessage());
                                collectedServerReply.add(message.receive().getMessage());
                                //System.out.println("current stack size: " + messages.size());
                            }));
            clientThread.start();
            System.out.println("finished client thread");
        } catch (IOException e) {
            logger.error("Exception occurred while reading from System.in. Exception: ", e);
        }
        messages.addAll(collectedServerReply);
    }
}
}

Client 类有一个setter，它接受任何实现ReplyListerner 接口(interface)的实例。后者作为一个抽象的服务器来收集回复数据:

public interface ReplyListener {

    void onReply(Message message) throws IOException;
}

我无法在 public void start(int numberOfThreads, List<String> messages) 方法中收集所有服务器回复.但是，当我简单地执行以下操作时:

 Thread clientThread = new Thread(
                        new ClientTask(port, ip)
                                .setInputStream(new FileInputStream("commands.txt"))
                                .setReplyListener(message -> {
                                    System.out.println(message.receive().toString());}));

我可以在我的 System.in 中看到相应的服务器回复

我在哪里犯了错误，如何实现收集服务器回复的目标？

编辑:

如何使用我的实现:

public static void main(String[] args) throws IOException, InterruptedException {
    List<String> collectedMessages = Collections.synchronizedList(new ArrayList<>());

    Thread serverThread = new Thread(new ServerLauncher(new File("config.xml")));
    serverThread.setName("server_thread");
    serverThread.start();
    new Client(new File("config.xml"), new FileInputStream("commands.txt")).start(1, collectedMessages);
    System.out.println("Collected output ...");
    for (int i = 0; i < collectedMessages.size(); i++) {
        System.out.println(collectedMessages.get(i));
    }
}

Answer 1

如果您使用 List 的非同步实现，您可能会遇到竞争条件。 .

要么显式同步:

synchronized(messages)
    messages.addAll(collectedServerReply);
}

或者确保使用同步列表调用您的启动函数，例如 Vector
编辑 - 跟随你的。确保在显示列表之前等待所有线程完成。

clientThread.join();

当然，在第二个循环中执行此操作，否则您将失去多线程的所有优势。