对象的多线程数组列表

Question

我的程序有一个网站数组列表，我通过图像处理进行 I/O，从网站抓取数据并更新/插入数据库。现在它很慢，因为所有的 I/O 都在完成。我想通过允许我的程序使用线程运行来加快速度。列表中没有任何内容被删除，列表中的每个网站都是相互独立的，所以对我来说，让实例同时循环遍历列表似乎可以加快速度。

假设我的列表是 10 个网站，现在它当然会遍历位置 0 到 9，直到我的程序完成对所有网站的处理。

假设我想让 3 个线程同时循环遍历这个包含 10 个网站的列表，同时在各自独立的空间中执行所有 I/O 和数据库更新，但使用相同的列表。

website.get(0) // thread1
website.get(1) // thread2
website.get(2) // thread3

然后说如果thread2到达循环的末尾，它首先返回并在下一个位置工作

website.get(3) // thread2

然后thread3完成，获取下一个位置

website.get(4) // thread3

然后thread1最终完成并在下一个位置工作

website.get(5) // thread1

等，直到完成。这很容易设置吗？在某个地方我可以找到一个很好的例子吗？我在网上寻找其他地方谈论我的场景，但没有找到。

Answer 1

在我的应用程序中，我像这样使用ExecutorService，效果很好：

主要代码：

ExecutorService pool = Executors.newFixedThreadPool(3); //number of concurrent threads

for (String name : website) { //Your ArrayList
    pool.submit(new DownloadTask(name, toPath));
}

pool.shutdown();
pool.awaitTermination(5, TimeUnit.SECONDS); //Wait for all the threads to finish, adjust as needed.

你工作的实际班级：

private static class DownloadTask implements Runnable {

    private String name;
    private final String toPath;

    public DownloadTask(String name, String toPath) {
        this.name = name;
        this.toPath = toPath;
    }

    @Override
    public void run() {
        //Do your parsing / downloading / etc. here.

    }
}

一些注意事项：

• 如果您使用的是数据库，则必须确保没有两个线程同时写入该数据库。

请参阅here 了解更多信息。

Answer 2

正如其他 cmets/answer 中提到的，您只需要一个具有固定大小的线程池执行器（根据您的示例说 3 个），它运行 3 个线程，这些线程在同一个列表上进行迭代，而不会拾取重复的网站。

因此，除了线程池执行器之外，您可能还需要正确计算每个线程中的下一个索引以从该列表中选择元素，这样线程不会从列表中选择相同的元素，也不会错过任何元素。

因此，我认为您可以使用 BlockingQueue 而不是 list，这消除了索引计算部分并保证从集合中正确选择元素。

public class WebsitesHandler {

    public static void main(String[] args) {
        BlockingQueue<Object> websites = new LinkedBlockingQueue<>();
        ExecutorService executorService = Executors.newFixedThreadPool(3);
        Worker[] workers = new Worker[3];
        for (int i = 0; i < workers.length; i++) {
            workers[i] = new Worker(websites);
        }
        try {
            executorService.invokeAll(Arrays.asList(workers));
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private static class Worker implements Callable {

        private BlockingQueue<Object> websites;

        public Worker(BlockingQueue<Object> websites) {
            this.websites = websites;
        }

        public String call() {
            try {
                Object website;
                while ((website = websites.poll(1, TimeUnit.SECONDS)) != null) {
                    // execute the task
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "done";
        }
    }
}

Answer 3

我认为您需要使用最新版本的 java 更新自己，即 Java8

学习一下Streams API，一定能解决你的问题