标签: java.util.concurrent

我有一些可以轻松并行化的工作，并且我想使用 Java 线程在我的四核机器上分配工作。这是一种应用于旅行商问题的遗传算法。听起来不容易并行，但第一个循环很容易并行。我谈论实际演变的第二部分可能会也可能不会，但我想知道我是否因为实现线程的方式而变慢，或者是否因为算法本身而变慢。

另外，如果有人对我应该如何实现我想做的事情有更好的想法，我将非常感激。

在 main() 中，我有这个：

 final ArrayBlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(numThreads*numIter);
 ThreadPoolExecutor tpool = new ThreadPoolExecutor(numThreads, numThreads, 10, TimeUnit.SECONDS, queue);
 barrier = new CyclicBarrier(numThreads);
 k.init(tpool);

我有一个在 init() 内部完成的循环，如下所示：

for (int i = 0; i < numCities; i++) {
    x[i] = rand.nextInt(width);
    y[i] = rand.nextInt(height);
}

我改成这样：

int errorCities = 0, stepCities = 0;
stepCities = numCities/numThreads;
errorCities = numCities - stepCities*numThreads;

// Split up work, assign to threads                                                                        
for (int i = 1; i <= numThreads; i++) { …

我的应用程序使用 Executor 为大量任务提供线程池。通过分析和基准测试，我确定，当每个核心有多个线程时，我的应用程序运行速度最快。一个好的启发式方法是从每个核心 4 个线程开始，不断变化，直到达到 >90% CPU 或 >90% RAM。

是否有可用的执行器可以开箱即用地执行此操作？要么自动为每个核心使用 N 个线程（而不仅仅是一个），或者理想情况下，根据 CPU 和 RAM 使用情况限制线程池大小？

如果失败 - 如何以编程方式确定核心数量？

在 Java 8 中，我在java.util.concurrent.ConcurrentMap接口上实现了一个包装器，特别是remove(Object key)方法。由于我需要在删除过程中进行更多检查，因此我需要使用以下命令来实现该功能

compute(K key, BiFunction<? super K, ? super V, ? extends V> remappingFunction)

然而，这里的问题是：我必须接受Object但将其传递给computeas K。对于泛型，我什至无法进行运行时检查，例如if (key instanceof K) ...

compute我明白为什么选择这样的签名；如果只是Object，万一计算需要创建一个新条目，它就不能简单地使用该密钥。但我不确定如何解决这个问题 - 是否有任何推荐的模式，但使用多个调用，如下所示？

for (;;) {
    V v = map.get();
    if (check(k, v)) {
        if (map.remove(k, v)) return true;
    } else return false;
}

谢谢

FutureTask 包装器比简单的 Callable/Runnables 提供什么？我见过一些人以这种方式使用 future，但我不确定它到底给游戏带来了什么。

Callable<Integer> myComputation = () -> {return 0;};
FutureTask<Integer> task = new FutureTask<Integer>(myComputation);

//Why this...       
Executors.newSingleThreadExecutor().execute(task);
task.get();

//...over the conventional approach?
Future<Integer> future = Executors.newSingleThreadExecutor().submit(myComputation);
future.get();

我有以下程序，我正在使用java.util.concurrent.CountDownLatch和不使用await()方法它工作正常。

我是并发的新手，想知道await(). 在CyclicBarrier我可以理解为什么await()需要，但为什么在CountDownLatch？

班级CountDownLatchSimple：

public static void main(String args[]) {
  CountDownLatch latch = new CountDownLatch(3);
  Thread one = new Thread(new Runner(latch),"one");
  Thread two = new Thread(new Runner(latch), "two");
  Thread three = new Thread(new Runner(latch), "three");

  // Starting all the threads
  one.start(); two.start(); three.start();
  
}

类Runner实现Runnable：

CountDownLatch latch;

public Runner(CountDownLatch latch) {
    this.latch = latch;
}

@Override
public void run() {
    System.out.println(Thread.currentThread().getName()+" is …

我正在使用 Java 8 并想知道computeIfPresent操作是ConcurrentHashMap锁定整个表/地图还是仅锁定包含密钥的 bin。

从该方法的文档中computeIfPresent：

其他线程在此映射上尝试的一些更新操作可能会在计算过程中被阻塞，因此计算应该简短且简单，并且不得尝试更新此映射的任何其他映射

当为一个键调用这个方法时，这看起来整个地图都被锁定了。如果更新某个键的值，为什么必须锁定整个地图？锁定包含键/值对的 bin 不是更好吗？

我正在向基于 Jersey 的 Web 服务添加一个新端点。支持端点的逻辑需要对另一个服务进行 10 到 50 次调用。这些调用是独立的并且可以并行化，因此我正在考虑使用执行器服务来跨多个线程分配工作。

我想知道是否应该为每个请求实例化一个 executorService 或者是否应该在 web 应用程序中存在一个共享的 executorService 实例。在后一种情况下，我将如何决定它应该运行的线程数？

我正在使用 JDK 9 Flow API 创建一个用户事件系统，所以我有一个房间（它实现了Flow.Subscriber<Notification>），它可能有很多用户，每个用户都可以随时提供（调度）更新。

当用户进入房间时，我订阅房间的更新user.subscribe(this)。但是没有退订，当用户离开房间时我如何退订？

public abstract class Room implements Flow.Subscriber<Notification> {
    private Flow.Subscription subscription;

    public void addUser(User user) {
        user.subscribe(this);
    }

    public void removeUser(User user) {
        // How can I unsubscribe the user?
    }

    @Override
    public void onSubscribe(final Flow.Subscription subscription) {
        this.subscription = subscription;
        subscription.request(1);
    }

    @Override
    public void onError(final Throwable throwable) {
        // ...
    }

    @Override
    public void onNext(final Notification notification) {
        // ...
        subscription.request(1);
    }

    @Override
    public void onComplete() {
        // …

我发现我们可以使用newKeySet();或 withkeySet(default value)从ConcurrentHashMap. 当写操作超过读操作时，这是创建线程安全集的最佳方法吗？

我读到 CopyOnWriteArraySet当阅读比写作更多时，似乎更好。

欢迎所有可以帮助我们了解更多信息的答案。

在转移方法中，判断扩展终止（或帮助转移线程完成）的条件是if (i < 0 || i >= n || i + n >= nextn) {。我知道i < 0这个条件意味着所有的 bin 都被分配了，但我不明白其他两个条件的含义： i >= n和i + n >= nextn

是否i >= n考虑数据溢出？(-2147483648 - 1 = 2147483647);

是i + n >= nextn一样的i >= n吗？（我不这么认为）

private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
        int n = tab.length, stride;
        //...
        int nextn = nextTab.length;
        ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
        boolean advance = true;
        boolean …