对于大数据量关联的业务处理逻辑,比较直接的想法就是用JDK提供的并发包去解决多线程情况下的业务数据处理。线程池可以提供很好的管理线程的方式,并且可以提高线程利用率,并发包中的原子计数在多线程的情况下可以让我们避免去写一些同步代码。
这里就先把jdk并发包中的线程池处理器ThreadPoolExecutor 以原子计数类AomicInteger 和倒数计时锁CountDownLatch的一些常用api介绍下。
一、ThreadPoolExecutor
JDK建议我们用工厂方法来创建线程池。Executors有四种工厂方法
1
、Executors.newCachedThreadPool
2
、Executors.newFixedThreadPool
3
、Executors.newSingleThreadExecutor
4
、Executors.newScheduledThreadPool()
当然也可以自己订制线程池来管理线程。下面简单举个订制线程池的构造器例子:
ThreadPoolExecutor threadPool = new ThreadPoolExecutor(ThreadPoolUtil.CORE_POOL_SIZE, ThreadPoolUtil.MAX_POOL_SIZE, ThreadPoolUtil.KEEP_ALIVE_TIME, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(ThreadPoolUtil.WORK_QUEUE_SIZE), new ThreadPoolExecutor.CallerRunsPolicy());
这里结合execute的实现来简单解释一下参数的含义。
1、corePoolSize:核心线程的数目。也就是线程池维护的最小线程数目。如果Idle time超过了keepAliveTime,线程实例就会被终止,移出线程池。对于核心线程数目的大小需要根据不同的应用场景和CPU的情况来具体确定,以保证最佳的线程数。
2、maximumPoolSize:线程池维护的最大线程数目。对于无界队列,这个是形同虚设的。在有界队列中可以通过该项设置特殊的handler来处理线程。
3、keepAliveTime 以及unit是用来控制idleTime的。即线程池维护线程所允许的空闲时间以及时间单位。
4、
workQueue任务队列:有多种队列可供选择,SynchronousQueue
:直接提交类型的队列。不保持线程而是直接创建。LinkedBlockingQueue:无界队列,维持线程数目在corePoolSize之内不会创建新的线程,而是放到队列中,适合任务独立的线程类型。ArrayBlockingQueue:有界队列。有助于防止资源耗尽但是需要调整好合适的队列size以及最大线程数目。
5、policy:
处理策略。也就是当任务数超过
maximumPoolSize时的处理策略。可以自己定义单独的handler也可以用线程池提供的。
AbortPolicy:抛出java.util.concurrent.RejectedExecutionException异常
CallerRunsPolicy:重试添加当前的任务,他会自动重复调用execute()方法
DiscardOldestPolicy:抛弃旧的任务
DiscardPolicy:抛弃当前的任务
通过以上几点可以看一下execute方法的内部。其处理的顺序基本上是corePoolSize会先填满,然后任务会放到队列中,有空闲的线程就会将队列中的任务“offer”出来执行,当队列也满了的时候就会交给handler来处理任务。
public void execute(Runnable command) {
//任务为空的时候抛出空
if (command == null)
throw new NullPointerException();
if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {
//添加线程到线程池中运行或者如果size>core放到队列中
if (runState == RUNNING && workQueue.offer(command)) {
if (runState != RUNNING || poolSize == 0)
ensureQueuedTaskHandled(command);
}
else if (!addIfUnderMaximumPoolSize(command))
//对于超出maxmum的线程,调用对应的reject策略
reject(command); // is shutdown or saturated
}
}
二、 AomicInteger计数器
在多线程的情况下写一个计数器用同步的方式比较麻烦。并发包提供的AomicInteger将程序员从写一些同步代码中解脱出来。AomicInteger中最重要的方法就是incrementAndGet()。看一下源码:
/**
* Atomically increments by one the current value.
*
* @return the updated value
*/
public final int incrementAndGet() {
for (;;) {
int current = get();
int next = current + 1;
if (compareAndSet(current, next))
return next;
}
}
其中compareAndSet方法是保证不会重复计数的关键。其中调用了CAS原语的操作,使得效率和准确性得以保证,但是起上一步在读next的时候如果另一个线程修改了当前值还是会出现问题,但这种问题的几率不高,而且也比通过synchronized和notify要更简便直接。每次调用incrementAndGet就会使AomicInteger的值加一返回,调用起来也很方便。
三、 CountDownLatch
可以配合AomicInteger一起进行倒数计时。其控制了指定数目线程全部执行完相应的逻辑后进行减数计算,知道数值为0之后唤醒后续线程继续执行。在这次应用中控制所有的AomicInteger计数结束后,也即所有子线程执行结束之后,再运行父线程来返回结果。CountDownLatch 中常用的两个方法就是await()和countDown()。顾名思义,await方法会阻塞线程往后执行,而去继续执行子线程。countDown()也是原语操作。每次调用都会将latch的数值减一,其初始数值可以在构造其中指定,一般是需执行线程的size。在调用的时候为了确保其能countDown,建议将其放到finally代码块中。保证计数与否都要countDown,要不父线程就永远被阻塞在那里了。countDown调用的内部静态类Sync继承的AbstractQueuedSynchronizer的releaseShared()方法。在tryReleaseShared(1)返回true的时候会进一步执行 doReleaseShared()方法。看一下其内部处理:
private void doReleaseShared() {
for (;;) {
Node h = head;
if (h != null && h != tail) {
int ws = h.waitStatus;
if (ws == Node.SIGNAL) {
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue; // loop to recheck cases
unparkSuccessor(h);
}
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // loop on failed CAS
}
if (h == head) // loop if head changed
break;
}
}
Node是用来标记等待状态。如果是head就保证等待状态不被取消。如果是tail表示是等待队列的尾元素。通过CAS原语操作来判断其在操作系统中的waitStatus。
相关推荐
并发包 同步容器类 Vector与ArrayList区别 1.ArrayList是最常用的List实现类,内部是通过数组实现的,它允许对元素进行快速随机访问。数组的缺点是每个元素之间不能有间隔,当数组大小不满足时需要增加存储能力,...
java并发包之Callable和Future java并发包之Callable和Future java并发包之Callable和Future java并发包之Callable和Future java并发包之Callable和Future java并发包之Callable和Future
java 并发包 java 并发包 java 并发包
Java并发包思维导图.xmind Java并发包思维导图.xmind Java并发包思维导图.xmind Java并发包思维导图.xmind Java并发包思维导图.xmind Java并发包思维导图.xmind
java并发包源码分析(3)ThreadLocal 详细配图讲解 java并发包源码分析(3)ThreadLocal 详细配图讲解
java并发包讲解 可以找我要代码,qq 3341386488 ## 线程安全-并发容器JUC--原理以及分析 1.arrayList --copyonWriteArraylist 优缺点 2.HashSet,TreeSet -- CopyONWriteArraySet,ConcurrentSkipListSet 3....
17. 线程池执行者 ThreadPoolExecutor 18. 定时执行者服务 ScheduledExecutorService 19. 使用 ForkJoinPool 进行分叉和合并 20. 锁 Lock 21. 读写锁 ReadWriteLock 22. 原子性布尔 AtomicBoolean 23. 原子性整型 ...
线程安全、volatile关键字、原子性、并发包、死锁、线程池学习笔记
ConcurrentSkipListMap等类的概念、原理、数据结构、使用场景描述。
并发包总结 思维导图 concurrent
Java 7并发包最新思维导图,基于《深入浅出Java concurrency》修改
Java分布式应用学习笔记04JDK的并发包的集合总结
CountDownLatch计数器闭锁是一个能...类似于一扇门:在闭锁到达结束状态之前,这扇门一直是关闭着的,不允许任何线程通过,当到达结束状态时,这扇门会打开并允许所有的线程通过。且当门打开了,就永远保持打开状态。
Java核心 JAVA多线程,集合,并发相关学习,日行一码
java.util.concurrent包提供了创建并发应用程序的工具,本资源主要是对该api进行详细的解读,并对api的使用做出安全高效的引用建议.
对象交换器:当两个线程处理的结果需要交换进行处理的时候,使用并发包中适合多线程情况下的数据结构:非阻塞队列:ConcurrentLinkedQueue Conc
第5章介绍Java并发包中与锁相关的API和组件,以及这些API和组件的使用方式与实现细节。 第6章介绍了Java中的大部分并发容器,并深入剖析其实现原理,让读者领略大师的设计技巧。 第7章介绍了Java中的原子操作类,并...
第5章介绍Java并发包中与锁相关的API和组件,以及这些API和组件的使用方式和实现细节。第6章介绍了Java中的大部分并发容器及其实现原理。第7章介绍了Java中的原子操作类,并给出一些实例。第8章介绍了Java中提供的很...
1、Java并发体系-第一阶段-多线程基础知识 2、Java并发体系-第二阶段-锁与同步-[1] ...5、Java并发包-第三阶段-JUC并发包-[1] 6、Java并发包-第三阶段-JUC并发包-[2] 7、Java并发体系-第四阶段-AQS源码解读-[1]
第5章介绍Java并发包中与锁相关的API和组件,以及这些API和组件的使用方式与实现细节。第6章介绍了Java中的大部分并发容器,并深入剖析其实现原理,让读者领略大师的设计技巧。第7章介绍了Java中的原子操作类,并给...