如何以多线程方式调用返回列表元素的方法

Question

我有这个方法：

    public List<String> composeList (DataBaseObject dBO) {
    List<String> valueList = new ArrayList<>();
       for (String separatedFieldName : separatedFieldNames) {
           object = PropertyUtils.getProperty(object, separatedFieldName);
           valueList.add(object.toString());
       }                        
    }

我有一个包含 1000 个dBO 对象的列表，并希望以多线程方式调用此方法。

但是这个方法的返回也会进入一个列表

这里是调用者：

List<List<String>> valueLists =  new ArrayList<>();
    for (DataBaseObject dBO : listOfDBOs)
    valueLists.add(composeList(dBOObject));

由于现在的机器有多个内核，我想知道如何使用它们。就像我如何在并行中调用 composeList 并将结果存储在一个 ArrayList 中一样。

我知道我可以使用 Collections.SynchronizedList，但是 composeList 的执行时间非常短，以至于我最终会按顺序添加元素，即使是多线程的，它仍然是顺序执行，因为每次添加都会放锁定 Sysnchronized 列表。

这听起来像是一个设计问题，但它仍然与编程相关。对于这种情况，我非常感谢任何帮助。

Answer 1

可能，这个解决方案有点“传统风格”，没有任何酷炫的新东西，比如 stream 等，但我会做如下的事情：

public class AnyClass {

    private static final AtomicInteger index = new AtomicInteger(0);
    private static final Object lock = new Object();      

    public class ComposerThread implements Runnable {

        private List<DataBaseObject> dboList;
        private List<List<String>> valueList;
        private List<String> fieldNames;

        public MyThread(List<DataBaseObject> dboList, List<List<String>> valueList, 
                        List<String> fieldNames) {
            this.valueList= valueList;
            this.dboList = dboList;
            this.fieldNames = fieldNames;
        }

        public void run() {
            int i = index.getAndIncrement(); //thread takes next unprocessed index
            while(i<dboList.size()){
                DataBaseObject object = dboList.get(i);
                List<String> list = new ArrayList<>();
                for (String separatedFieldName : fieldNames) {
                    Object object = PropertyUtils.getProperty(object, separatedFieldName);
                    list.add(object.toString());
                }  

                synchronized (lock) {  //addition synchronized
                    valueList.add(list);
                }
                i = index.getAndIncrement(); //thread takes next unprocessed index
            }
        }
    }
    ....
}

注意，AtomicInteger index 和 Object lock 必须是 final static，因为在同步中使用。现在您可以在同一个父类中使用ComposerThread 内部类：

public class AnyClass{

    ....

    private List<List<String>> composeValueList(List<DataBaseObject> dboList,
                                                List<String> fieldNames, int threadCount) {

        index.set(0);//reset index before process dboList
        List<List<String>> valueList = new List<List<String>>();

        Thread [] pool = new Thread[threadCount];
        for(int i=0; i<pool.length; i++){
            pool[i] = new Thread(new ComposerThread (objectList, valueList, fieldNames));
            pool[i].start();
        }
        for(Thread thread : threadPool){
            thread.join(); //just wait while all will be done
        }
        return valueList;
    }
}

如您所见，您甚至可以设置线程数。

Answer 2

Java 8 并行流正是为这种情况而设计的。

List<String> dbFieldValues = dbObjectList.parallelStream()
    .flatMap(seperatedFieldNames().parallelStream()
        .map(fn -> PropertyUtils.getProperty(db, fn).toString()))
    .collect(Collectors.toList());

假设seperatedFieldNames 的集合支持并行流（例如ArrayList），这将使用多个线程，而无需自己创建它们。

请注意，这是假设 getProperty 没有副作用。

Answer 3

阅读 Java 中的 ExecutorService 和 ForkJoin 框架。它可以帮助你实现你想要的。

谢谢。