这是一个java并发问题。需要完成 10 个作业,每个作业有 32 个工作线程。工作线程会增加一个计数器。一旦计数器达到 32,就意味着这项工作完成,然后清理计数器映射。从控制台输出来看,我预计将输出 10 个“done”,池大小为 0,counterThread 大小为 0。
问题是:
大多数时候,“池大小:0 和 countThreadMap 大小:3”将是 打印出来。即使所有线程都消失了,但 3 个工作却没有 还没完。
有时候,我可以在第27行看到nullpointerException。我已经使用了ConcurrentHashMap和AtomicLong,为什么仍然有并发 异常(exception)。
谢谢
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicLong;
public class Test {
final ConcurrentHashMap<Long, AtomicLong[]> countThreadMap = new ConcurrentHashMap<Long, AtomicLong[]>();
final ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
final ThreadPoolExecutor tPoolExecutor = ((ThreadPoolExecutor) cachedThreadPool);
public void doJob(final Long batchIterationTime) {
for (int i = 0; i < 32; i++) {
Thread workerThread = new Thread(new Runnable() {
@Override
public void run() {
if (countThreadMap.get(batchIterationTime) == null) {
AtomicLong[] atomicThreadCountArr = new AtomicLong[2];
atomicThreadCountArr[0] = new AtomicLong(1);
atomicThreadCountArr[1] = new AtomicLong(System.currentTimeMillis()); //start up time
countThreadMap.put(batchIterationTime, atomicThreadCountArr);
} else {
AtomicLong[] atomicThreadCountArr = countThreadMap.get(batchIterationTime);
atomicThreadCountArr[0].getAndAdd(1);
countThreadMap.put(batchIterationTime, atomicThreadCountArr);
}
if (countThreadMap.get(batchIterationTime)[0].get() == 32) {
System.out.println("done");
countThreadMap.remove(batchIterationTime);
}
}
});
tPoolExecutor.execute(workerThread);
}
}
public void report(){
while(tPoolExecutor.getActiveCount() != 0){
//
}
System.out.println("pool size: "+ tPoolExecutor.getActiveCount() + " and countThreadMap size:"+countThreadMap.size());
}
public static void main(String[] args) throws Exception {
Test test = new Test();
for (int i = 0; i < 10; i++) {
Long batchIterationTime = System.currentTimeMillis();
test.doJob(batchIterationTime);
}
test.report();
System.out.println("All Jobs are done");
}
}
最佳答案
让我们深入探讨一下与线程相关的编程中一个人可能犯的所有错误:
Thread workerThread = new Thread(new Runnable() {
…
tPoolExecutor.execute(workerThread);
您创建一个线程
,但不启动它,而是将其提交给执行器。让 Thread
无缘无故地实现 Runnable
是 Java API 的一个历史性错误。现在,每个开发人员都应该意识到,没有理由将Thread
视为Runnable
。如果您不想start
手动创建线程,不要创建线程
。只需创建 Runnable
并将其传递给 execute
或 submit
.
我想强调后者,因为它返回一个 Future
,它免费为您提供您正在尝试实现的内容:任务完成时的信息。使用 invokeAll
更容易它将提交一堆 Callable 并在所有完成后返回。由于您没有告诉我们任何有关您的实际任务的信息,因此不清楚您是否可以让您的任务简单地实现 Callable (可能返回 null )而不是 Runnable .
如果你不能使用Callable
或者不想在提交后立即等待,你必须记住返回的Future
并在稍后查询它们时间:
static final ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
public static List<Future<?>> doJob(final Long batchIterationTime) {
final Random r=new Random();
List<Future<?>> list=new ArrayList<>(32);
for (int i = 0; i < 32; i++) {
Runnable job=new Runnable() {
public void run() {
// pretend to do something
LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(r.nextInt(10)));
}
};
list.add(cachedThreadPool.submit(job));
}
return list;
}
public static void main(String[] args) throws Exception {
Test test = new Test();
Map<Long,List<Future<?>>> map=new HashMap<>();
for (int i = 0; i < 10; i++) {
Long batchIterationTime = System.currentTimeMillis();
while(map.containsKey(batchIterationTime))
batchIterationTime++;
map.put(batchIterationTime,doJob(batchIterationTime));
}
// print some statistics, if you really need
int overAllDone=0, overallPending=0;
for(Map.Entry<Long,List<Future<?>>> e: map.entrySet()) {
int done=0, pending=0;
for(Future<?> f: e.getValue()) {
if(f.isDone()) done++;
else pending++;
}
System.out.println(e.getKey()+"\t"+done+" done, "+pending+" pending");
overAllDone+=done;
overallPending+=pending;
}
System.out.println("Total\t"+overAllDone+" done, "+overallPending+" pending");
// wait for the completion of all jobs
for(List<Future<?>> l: map.values())
for(Future<?> f: l)
f.get();
System.out.println("All Jobs are done");
}
但请注意,如果您不需要 ExecutorService
来执行后续任务,则等待所有作业完成会更容易:
cachedThreadPool.shutdown();
cachedThreadPool.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);
System.out.println("All Jobs are done");
<小时/>
但是,无论手动跟踪作业状态有多么不必要,让我们深入研究一下您的尝试,这样您就可以避免将来出现错误:
if (countThreadMap.get(batchIterationTime) == null) {
ConcurrentMap
是线程安全的,但这不会将并发代码转变为顺序代码(这会使多线程变得无用)。上面这一行可能会同时被最多 32 个线程处理,所有线程都会发现该键还不存在,因此可能会有多个线程将初始值放入映射中。
AtomicLong[] atomicThreadCountArr = new AtomicLong[2];
atomicThreadCountArr[0] = new AtomicLong(1);
atomicThreadCountArr[1] = new AtomicLong(System.currentTimeMillis());
countThreadMap.put(batchIterationTime, atomicThreadCountArr);
这就是为什么这被称为“检查然后行动”反模式的原因。如果多个线程要处理该代码,它们都会放入
它们的新值,并确信这是正确的事情,因为它们在执行操作之前已经检查了初始条件,但对于除一个线程之外的所有线程执行操作时条件发生了变化,并且它们正在覆盖之前的 put
操作的值。
} else {
AtomicLong[] atomicThreadCountArr = countThreadMap.get(batchIterationTime);
atomicThreadCountArr[0].getAndAdd(1);
countThreadMap.put(batchIterationTime, atomicThreadCountArr);
由于您正在修改已存储到映射中的 AtomicInteger
,因此 put
操作没有用,它会放入之前检索到的数组。如果没有出现如上所述的可以有多个初始值的错误,则 put
操作不会产生任何效果。
}
if (countThreadMap.get(batchIterationTime)[0].get() == 32) {
同样,使用ConcurrentMap
不会将多线程代码转换为顺序代码。虽然很明显,唯一的最后一个线程会将原子整数更新为 32
(当初始竞争条件未实现时),但不能保证所有其他线程都已通过此 if 语句。因此,不止一个,最多所有线程仍然可以在此时执行并看到
32
的值。或者……
System.out.println("done");
countThreadMap.remove(batchIterationTime);
看到 32
值的线程之一可能会执行此remove
操作。此时,可能仍然有线程没有执行上面的 if
语句,现在看不到值 32
但产生一个 NullPointerException
作为应该包含 AtomicInteger 的数组不再在 map 中。这就是偶尔会发生的事情……
关于Java 并发计数器未正确清理,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/31300120/