如何在 Java 中的循环中使用多线程

Question

这就是我想要做的。我在while loop 中记录来自不同传感器的数据，直到用户停止记录。我想每秒记录尽可能多的数据。传感器需要不同的时间来返回一个值，介于 200 毫秒和 3 秒之间。因此，依次调用传感器不是一种选择。

依次调用传感器如下所示：

List<DataRow> dataRows= new ArrayList<DataRow>();

while (recording) {
   DataRow dataRow = new DataRow();

   dataRow.setDataA(sensorA.readData());
   dataRow.setDataB(sensorB.readData());
   dataRow.setDataC(sensorC.readData());

   dataRows.add(dataRow);
}

根据传感器，读取数据看起来（非常简化）

public class SensorA {

   public SensorAData readData(){
      sensorA.startSensing();

      try {
          TimeUnit.MILLISECONDS.sleep(750);
      } catch (InterruptedException e) {
          Thread.currentThread().interrupt(); 
      }

      return sensorA.readAndConvertByteStream();
   }
}

要利用多线程，SensorA 能否在循环中实现Callable 和接收Future 对象？还是应该将while loop 放在实现接口Runnable 的run() 方法中？

基本上，即使循环已经至少进一步迭代，Java（或线程）是否可以写入正确的dataRow 对象？如果不是，如何解决这个问题？

Answer 1

如果我正确理解您的需求，这可能是您想要的解决方案：

• 在每次迭代中，n 个传感器由 n 个并发线程读取，
• 如果所有线程都收集了传感器数据，则将新结果行添加到列表中

工作代码：

public class TestX {

    private final ExecutorService pool = Executors.newFixedThreadPool(3);
    private final int N = 10;

    // all sensors are read sequentially and put in one row
    public void testSequential() {
        int total = 0;
        long t = System.currentTimeMillis();

        for (int i = 0; i < N; i++) {
            System.out.println("starting iteration " + i);

            int v1 = getSensorA();    // run in main thread
            int v2 = getSensorB();    // run in main thread
            int v3 = getSensorC();    // run in main thread

            // collection.add( record(v1, v2, v3)
            total += v1 + v2 + v3;
        }

        System.out.println("total = " + total + "   time = " + (System.currentTimeMillis() - t) + " ms");
    }

    // all sensors are read concurrently and then put in one row
    public void testParallel() throws ExecutionException, InterruptedException {
        int total = 0;
        long t = System.currentTimeMillis();

        final SensorCallable s1 = new SensorCallable(1);
        final SensorCallable s2 = new SensorCallable(3);
        final SensorCallable s3 = new SensorCallable(3);

        for (int i = 0; i < N; i++) {
            System.out.println("starting iteration " + i);

            Future<Integer> future1 = pool.submit(s1);  // run in thread 1
            Future<Integer> future2 = pool.submit(s2);  // run in thread 2
            Future<Integer> future3 = pool.submit(s3);  // run in thread 3

            int v1 = future1.get();
            int v2 = future2.get();
            int v3 = future3.get();

            // collection.add( record(v1, v2, v3)
            total += v1 + v2 + v3;
        }

        System.out.println("total = " + total + "   time = " + (System.currentTimeMillis() - t) + " ms");
    }

    private class SensorCallable implements Callable<Integer> {

        private final int sensorId;

        private SensorCallable(int sensorId) {
            this.sensorId = sensorId;
        }

        @Override
        public Integer call() throws Exception {
            switch (sensorId) {
                case 1: return getSensorA();
                case 2: return getSensorB();
                case 3: return getSensorC();
                default:
                    throw new IllegalArgumentException("Unknown sensor id: " + sensorId);
            }
        }
    }

    private int getSensorA() {
        sleep(700);
        return 1;
    }

    private int getSensorB() {
        sleep(500);
        return 2;
    }

    private int getSensorC() {
        sleep(900);
        return 2;
    }

    private void sleep(long ms) {
        try {
            Thread.sleep(ms);
        } catch (InterruptedException e) {
            // ignore
        }
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        new TestX().testSequential();
        new TestX().testParallel();
    }
}

和输出：

starting iteration 0
starting iteration 1
starting iteration 2
starting iteration 3
starting iteration 4
starting iteration 5
starting iteration 6
starting iteration 7
starting iteration 8
starting iteration 9
total = 50   time = 21014 ms

starting iteration 0
starting iteration 1
starting iteration 2
starting iteration 3
starting iteration 4
starting iteration 5
starting iteration 6
starting iteration 7
starting iteration 8
starting iteration 9
total = 50   time = 9009 ms

-- 编辑--

在 java 8 中，您可以使用方法引用来摆脱 Callable 类，只需编写：

Future<Integer> future1 = pool.submit( this::getSensorA() );
Future<Integer> future2 = pool.submit( this::getSensorB() );
Future<Integer> future3 = pool.submit( this::getSensorC() );