Java 并发数 : how to retrieve data from several thousands TCP sockets at once and synchronously?

Question

我是并发网络编程的新手。

我的全局任务:开发每 10 秒从 3000 个 tcp 网络设备同步获取温度的 java 应用程序。

获取温度过程/步骤:

• 通过 TCP 套接字连接到特定的 IP 地址(超时 3 秒)
• 将登录名/密码写入套接字
• 从 socket 读取温度
• 关闭连接

从一台设备检索温度大约需要 300-400 毫秒。 检索的总时间为 400ms * 3000 = 120 秒

我需要为每个任务启动 3000 个线程来同时检索或使用队列和线程池吗？

是否存在其他想法？

请把我带到正确的方向。

Answer 1

可以使用Java API，Java提供ScheduledThreadPoolExecutor这会让你:

• 创建一个线程池(并让您控制为此目的在 JVM 中需要多少个线程，并且 JVM 管理该线程池)。
• 公开指定任务时间表的方法。

我已经为你提供了一个示例实现，我添加了代码注释以便于你理解，你也可以引用在线文档或示例。如有任何问题，请随时告诉我。

顺便说一句，如果你想从线程的执行中得到结果，那么你可以使用 Callable 接口(interface)而不是使用 Runnable 接口(interface)，它可以让你返回一些东西， ScheduledThreadPoolExecutor 具有重载方法，可让您传递 Callable 对象。

示例代码:

import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class ScheduledThreadPoolExecutorExample {

    private final static int MAX_NUMBER_OF_THREADS = 2;
    private final static int EXECUTION_SCHEDULE_PERIOD_IN_SECONDS = 5;
    private final static int INITAL_DELAY_IN_SECONDS = 0;

    // set maximum number of threads as per your requirement/performance tuning, for testing set it to "2" and to have better feel.
    private final static ScheduledThreadPoolExecutor SCHEDULED_THREAD_POOL_EXECUTOR = new ScheduledThreadPoolExecutor(MAX_NUMBER_OF_THREADS);

    public static void main(String[] args) {
        scheduleTask(new MyRunnableTask("google1.com", 80)); // if you want more fields to set then either use constructor or have setter methods.
        scheduleTask(new MyRunnableTask("google2.com", 80)); // if you want more fields to set then either use constructor or have setter methods.
        scheduleTask(new MyRunnableTask("google3.com", 80)); // if you want more fields to set then either use constructor or have setter methods.
        scheduleTask(new MyRunnableTask("google4.com", 80)); // if you want more fields to set then either use constructor or have setter methods.
    }

    private static void scheduleTask(Runnable runnable) {
        SCHEDULED_THREAD_POOL_EXECUTOR.scheduleAtFixedRate(runnable, INITAL_DELAY_IN_SECONDS, EXECUTION_SCHEDULE_PERIOD_IN_SECONDS, TimeUnit.SECONDS);
    }
}

可运行:

import java.util.Date;

public class MyRunnableTask implements Runnable {

    private String hostName;
    private int port;

    MyRunnableTask(String _hostName, int _port){
        this.hostName = _hostName;
        this.port = _port;
    }

    @Override
    public void run() {
        System.out.println(this.hostName + ":: I am getting executed: " + this.hashCode() + " | " + Thread.currentThread().getId() + " | " + new Date());

        // implement your socket programming code here

        try {
            Thread.sleep(500);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}