标签: java.util.concurrent

ScheduledExecutorService从ExecutorService,shutdown()和shutdownNow()继承两个方法.他们之间的区别:

shutdown启动有序关闭,其中先前提交的任务被执行,但不会接受任何新任务.如果已经关闭,调用没有其他影响.

shutdownNow 尝试停止所有正在执行的任务,停止等待任务的处理,并返回等待执行的任务列表.

现在我想暂停等待任务的处理,而我不想中断当前正在执行的任务.我不能打断线程,因为涉及第三方库并且它们不能很好地处理中断:-(但是我需要取消当前没有执行的计划任务,因为大多数都是在一小时左右的时间内安排的.

处理这个问题的最佳方法是什么？我有什么选择？

什么是ReentrantLock#tryLock(long,TimeUnit)实现在尝试获取锁时执行的操作？假设线程A实际上拥有Lock of myLock和Thread B调用myLock.tryLock(10,SECONDS),是线程B正在休眠还是在等待？

换句话说,这是两个实现的区别:

1.

while (true)
   try {
     if (readLock.tryLock())
       return;
     MILLISECONDS.sleep(5);
   }catch (InterruptedException e) {}

2.

 while (true)
   try {
     if (readLock.tryLock(5,MILLISECONDS))
       return;
   }catch (InterruptedException e) {}

据我所知,链表和数组都可以无限制地增长,或者我错了吗？但是当我查看Executor Service中的文档时,我看到了:

无限的队列.使用无界队列(例如,没有预定义容量的LinkedBlockingQueue)将导致新任务在所有corePoolSize线程忙时在队列中等待.因此,只会创建corePoolSize线程.(而且maximumPoolSize的值因此没有任何影响.)

Unbounded Queue当LinkedBlockingQueue具有已定义的容量时,属性是否会发生变化？

这写给ArrayBlockingQueue:

有界的队列.有限队列(例如,ArrayBlockingQueue)在与有限maximumPoolSizes一起使用时有助于防止资源耗尽,但可能更难以调整和控制.队列大小和最大池大小可以相互交换:使用大型队列和小型池最小化CPU使用率,OS资源和上下文切换开销,但可能导致人为的低吞吐量.如果任务经常阻塞(例如,如果它们是I/O绑定的),系统可能能够为您提供比您允许的更多线程的时间.使用小队列通常需要更大的池大小,这会使CPU更加繁忙,但可能会遇到不可接受的调度开销,这也会降低吞吐量.

这个问题直接来自我在SO的上一个问题.我认为我的第二个问题的答案是否定的.所以我想理解为什么java.util.concurrent包中没有ConcurrentLinkedHashMap？我的意思是有一个ConcurrentHashMap但没有ConcurrentLinkedHashMap.在Concurrent环境中拥有这样的类是否完全没有意义？我的意思是它的非可用性的主要技术原因是什么？在Guava/Apache Commons中有类似的东西吗？

我有一组Futures通过提交Callables来创建Executor.伪代码:

for all tasks
  futures.add(executor.submit(new callable(task)))

现在我想让所有期货等待最多n秒,直到全部完成.我知道我可以打电话Future#get(timeout)但是如果我在循环中按顺序呼叫我的所有未来,则时间开始加起来.伪代码:

for all futures
  future.get(timeout)

get超时的块直到结果准备好.因此,如果第一个在超时之前完成,第二个也在超时之前完成,那么整个执行时间number of futures * timeout最多不是timeout.

因此,我正在寻找一种接受Futures 列表和超时的方法,并行运行,然后返回未来结果的集合.有任何想法吗？

我正在运行一个高度并发的Java程序.虽然许多线程正在向执行程序服务提交任务,但在某个点上主线程会调用ExecutorService.shutdownNow().

在这个动作之后,我希望:

1. 执行程序服务不接受任何其他任务
2. 执行程序服务的队列是清楚的
3. 其余正在运行的工作程序被中断,也就是说,如果他们正确管理InterruptedException和/或明确检查,他们会在某一时刻关闭 Thread.currentThread().isInterrupted()

由于我处于以下情况:

1. 执行程序服务由于a而"关闭" ExecutorService.shutdownNow(),但没有关闭,即ExecutorService.awaitTermination(long, TimeUnit)永远不会返回true
2. 有一些挂起的线程由我的执行器服务管理等待a BlockingQueue.take()
3. 如果我再次调用ExecutorService.shutdownNow(),挂起线程死于InterruptedException上BlockingQueue.take()

我想在调用之前这些线程已经收到BlockingQueue.take()了中断,并且忽略了InterruptedException.

我也想知道是否ExecutorService.shutdownNow()是线程安全的,即,即使线程池正在接收许多提交,它也能正常工作.

总而言之,我有两个问题:

1. 有没有其他方案可以解释我的情况？
2. 是否可能ExecutorService.shutdownNow()不是线程安全的？

在JDK 1.6中,Doug Lea final在该next领域之前使用.

static final class HashEntry<K,V> {
    final K key;
    final int hash;
    volatile V value;
    final HashEntry<K,V> next;

而在JDK 1.7中,next字段前面是volatile.我还注意到在JDK 1.7中,该get方法采用getObjectVolatile读取value字段的方法,该字段具有易失性加载语义.

我不知道Doug Lea之前使用过什么final.如果正确性存在问题,那么如何volatile在JDK 1.7(也是JDK 1.8)中替换它？

编辑:

具体来说,我的问题是我们可以final用volatileJDK 1.6的实现代替吗？

我想使用保存一些结果的并发哈希映射，

ConcurrentHashMap<Long,AtomicInteger>

如果键不存在，则添加一个新条目，或者通过键获取值并递增，如下所示：

if(map.contains(key))
  map.get(key).addAndGet(1);
else
  map.put(key,new AtomicInteger(1));    

put 操作不是安全的，如何解决这个问题？put 操作应该在同步块内吗？

我想了解是否volatile需要发布不可变对象.

例如,假设我们有一个不可变对象A:

// class A is immutable
class A {
  final int field1;
  final int field2;

  public A(int f1, int f2) {
    field1 = f1;
    field2 = f2;
  }
}

然后我们有一个B从不同线程访问的类.它包含对类对象的引用A:

// class B publishes object of class A through a public filed
class B {
  private /* volatile? */ A toShare;

  // this getter might be called from different threads
  public A getA(){
    return toShare;
  }

  // this might be called …

我正在尝试支持定期清除的哈希图上的并发性。我有一个缓存，可以存储一段时间的数据。每 5 分钟后，此缓存中的数据将发送到服务器。一旦我刷新，我想清除缓存。问题是当我刷新时，当我使用现有密钥执行此操作时，数据可能会写入此映射。我将如何使这个进程线程安全？

data class A(val a: AtomicLong, val b: AtomicLong) {
   fun changeA() {
      a.incrementAndGet()
   }
}

class Flusher {
   private val cache: Map<String, A> = ConcurrentHashMap()
   private val lock = Any()
   fun retrieveA(key: String){
       synchronized(lock) {
          return cache.getOrPut(key) { A(key, 1) }
       }
   }
 
   fun flush() {
      synchronized(lock) {
           // send data to network request
           cache.clear()
      }
   }
}

// Existence of multiple classes like CacheChanger
class CacheChanger{
  fun incrementData(){
      flusher.retrieveA("x").changeA()
  }
}

我担心上面的缓存没有正确同步。有没有更好/正确的方法来锁定这个缓存，这样我就不会丢失数据？我应该创建缓存的深层副本并清除它吗？

既然上面的数据可能被另一个更改器更改，那会不会导致问题？