0

我正在尝试维护多个线程之间的项目列表，每个线程一个（例如，每个线程一个套接字连接）。我将这个列表保存在一个ArrayDeque<>. 我面临的问题ArrayDeque<>是超过没有的项目。线程池中的线程数。

这是我的代码：

package com.practice;

import java.util.ArrayDeque;
import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class CompletableFutureApp implements AutoCloseable {
    private static void sleep(long timeMS) {
        try {
            Thread.sleep(timeMS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        try (CompletableFutureApp completableFutureApp = new CompletableFutureApp()) {

            Runnable[] tasksList1 = new Runnable[100];

            for (int i = 0; i < tasksList1.length; i++) {
                String msg1 = "TaskList 1 no.: " + i;

                tasksList1[i] = () -> {
                    //System.out.println(msg1);
                    sleep(300);
                };
            }

            completableFutureApp
                    .executeTasks(tasksList1)
                    .thenRun(() -> System.out.println("All tasks completed successfully"));
        }
    }

    private ExecutorService threadPool = Executors.newWorkStealingPool(10);
    private ArrayDeque<Integer> itemsAvailable = new ArrayDeque<>();

    private CompletableFuture executeTasks(Runnable... tasks) {
        CompletableFuture[] futures = new CompletableFuture[tasks.length];

        for (int i = 0; i < tasks.length; i++) {
            Runnable task = tasks[i];

            futures[i] = CompletableFuture.runAsync(() -> {
                Integer item = itemsAvailable.poll();
                if (item == null)
                    item = new Random().nextInt();

                task.run();

                itemsAvailable.add(item);

                System.out.println("Items available: " + itemsAvailable.size());
            }, threadPool);
        }

        return CompletableFuture.allOf(futures).thenRun(() -> System.out.println("Items available at last: " + itemsAvailable.size()));
    }

    @Override
    public void close() {
        threadPool.shutdown();
        try {
            threadPool.awaitTermination(100, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

在我的代码中，我正在创建一个大小为 10 的工作窃取池，谁能告诉我为什么不。在这种情况下，元素的ArrayDeque<>数量超过 10？

4

1 回答 1

4

从 Javadoc：

[Array deques] 不是线程安全的；在没有外部同步的情况下，它们不支持多线程并发访问。

使用外部同步，或选择专门为并发访问设计的不同类。

于 2019-07-30T11:43:31.627 回答