侯体宗的博客
  • 首页
  • Hyperf版
  • beego仿版
  • 人生(杂谈)
  • 技术
  • 关于我
  • 更多分类
    • 文件下载
    • 文字修仙
    • 群聊
    • 九宫格抽奖
    • 拼图
    • 消消乐
    • 相册

Java 并发包之线程池和原子计数

Java  /  管理员 发布于 3年前   378

对于大数据量关联的业务处理逻辑,比较直接的想法就是用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。



  • 上一条:
    Java实现的基于模板的网页结构化信息精准抽取组件:HtmlExtractor
    下一条:
    java Illegal overloaded getter method with ambiguous type for propert的解决
  • 昵称:

    邮箱:

    0条评论 (评论内容有缓存机制,请悉知!)
    最新最热
    • 分类目录
    • 人生(杂谈)
    • 技术
    • linux
    • Java
    • php
    • 框架(架构)
    • 前端
    • ThinkPHP
    • 数据库
    • 微信(小程序)
    • Laravel
    • Redis
    • Docker
    • Go
    • swoole
    • Windows
    • Python
    • 苹果(mac/ios)
    • 相关文章
    • java 正则表达式基础,实例学习资料收集大全 原创(0个评论)
    • java正则表达式彻底研究(0个评论)
    • java正则表达式验证函数(0个评论)
    • MVC、MVP和MVVM分别是什么(0个评论)
    • java 单例模式(饿汉模式与懒汉模式)(0个评论)
    • 近期文章
    • 如何优雅处理async await错误推荐:await-to-js库(0个评论)
    • lodash工具库(0个评论)
    • 在Laravel项目中使用中间件方式统计用户在线时长功能代码示例(0个评论)
    • 在Laravel中构建业务流程模型(0个评论)
    • windows系统中安装FFMpeg及在phpstudy环境php7.3 + php-ffmpeg扩展的使用流程步骤(0个评论)
    • 在go语言中对浮点的数组、切片(slice)进行正向排序和反向排序(0个评论)
    • 在go语言中对整数数组、切片(slice)进行排序和反向排序(0个评论)
    • 在go语言中对字符串数组、切片(slice)进行排序和反向排序(0个评论)
    • 最新国内免注册ChatGPT体验站_ChatGPT镜像站访问链接地址2023/3/28持续更新(0个评论)
    • 在Laravel项目中的实现无密码认证之:发送邮箱链接授权(0个评论)
    • 近期评论
    • 博主 在

      2023年国务院办公厅春节放假通知:1月21日起休7天中评论 @ xiaoB 你只管努力,剩下的叫给天意;天若有情天亦老,..
    • xiaoB 在

      2023年国务院办公厅春节放假通知:1月21日起休7天中评论 会不会春节放假后又阳一次?..
    • BUG4 在

      你翻墙过吗?国内使用vpn翻墙可能会被网警抓,你需了解的事中评论 不是吧?..
    • 博主 在

      go语言+beego框架中获取get,post请求的所有参数中评论 @ t1  直接在router.go文件中配就ok..
    • Jade 在

      如何在MySQL查询中获得当月记录中评论 Dear zongscan.com team, We can skyroc..
    • 2016-11
    • 2018-03
    • 2020-03
    Top

    Copyright·© 2019 侯体宗版权所有· 粤ICP备20027696号 PHP交流群

    侯体宗的博客