java - 如何在java中多线程查询数据50ms

Question

我将从机器中查询数据，每次查询 50 毫秒。我有一张带有 id 的 map 供查询。对于一个 id，我创建一个线程。最后我有 500 个线程。

for (Map.Entry<String, NodeId[]> entry1 : nodeIds.entrySet()) {
    try {
        NodeId[] entrNodeIds = entry1.getValue();

        new Thread() {
            public void run() {
                new ReadRegisteredValues().getValue(entry1.getKey(), entrNodeIds[0], client);
            };
        }.start();
    } catch (Exception e) {
        fileLogger.error("failed read node: " + entry1.getKey());
    }
}

此后，我使用 while(true) 每 50 毫秒读取每个线程中的值，如下所示:

public void getValue(String nodeName, NodeId registeredNodeId, UaClient client) {

    while (true) {
        try {
            DataValue value = null;
            try {
                value = client.readAttribute(registeredNodeId, Attributes.Value);
            } catch (ServiceException e1) {
                e1.printStackTrace();
            } catch (StatusException e1) {
                e1.printStackTrace();
            }

            TeocModel curTeocModel = new TeocModel(value.getSourceTimestamp().getTimeInMillis(), nodeName, value);
            KafkaStreamer.getInstance().startStreaming(curTeocModel.toJson());

            Thread.sleep(50);
        } catch (InterruptedException e) {
            fileLogger.error(e.getMessage());
        }
    }
}

但在这种情况下，我读取的值不是平行的。这是并行查询数据的最佳方法吗？

有什么想法吗？

Answer 1

您可以使用 ScheduledExecutorService 来为您进行调度。

ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(parallelism);
for (Map.Entry<String, NodeId[]> entry1 : nodeIds.entrySet()) {
    Runnable checkNode = () -> getValue(entry1.getKey(), entrNodeIds[0], client);
    scheduler.scheduleAtFixedRate(checkNode, 50, 50, ChronoUnit.MILLIS);
}

为此删除 getValue 中的 while 循环和 Thread.sleep()。

public void getValue(String nodeName, NodeId registeredNodeId, UaClient client) {

    try {
        DataValue value = null;
        try {
            value = client.readAttribute(registeredNodeId, Attributes.Value);
        } catch (ServiceException | StatusException e1) {
            e1.printStackTrace();
        }

        TeocModel curTeocModel = new TeocModel(value.getSourceTimestamp().getTimeInMillis(), nodeName, value);
        KafkaStreamer.getInstance().startStreaming(curTeocModel.toJson());

    } catch (Exception e) {
        fileLogger.error(e.getMessage());
    }
}

现在，更新将从其中一个线程每 50 毫秒执行一次。请注意，如果更新时间太长并且更新次数太多，执行队列将永远增长，最终您将遇到某种资源问题。