Skip to content

服务器托管,北京服务器托管,服务器租用-价格及机房咨询

Menu
  • 首页
  • 关于我们
  • 新闻资讯
  • 数据中心
  • 服务器托管
  • 服务器租用
  • 机房租用
  • 支持中心
  • 解决方案
  • 联系我们
Menu

java线程池最佳实践

Posted on 2023年10月4日2023年10月4日 by hackdl

线程池工具类

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

// 线程池构建器 模板用法参见 common.ThreadPoolBuilderTest
// 队列满了之后才会创建第(corePoolSize+1)个线程, 而LinkedBlockingQueue 默认大小为int.Max,SynchronousQueue 大小为1
// 默认队列满之后的拒绝策略是抛出异常, 会导致任务线程停止并且pool.shutdown()永远不能正常结束
// 必须捕获任务线程的异常
// 线程按顺序消费队列
public final class ThreadPoolUtil {

    private static final Logger log = LoggerFactory.getLogger(ThreadPoolUtil.class);

    // cpu核心数
    public static final int cpu_num = Runtime.getRuntime().availableProcessors();

    public static ThreadPoolExecutor buildPool(String name) {
        return buildPool(name, cpu_num);
    }

    /**
     * 构建线程池
     *
     * @param threadNamePrefix 任务线程名字前缀
     * @param maxSize          线程池最大线程数
     * @return 线程池
     */
    public static ThreadPoolExecutor buildPool(String threadNamePrefix, int maxSize) {
        // 创建固定线程数,任务队列无穷大的线程池
        ThreadPoolExecutor pool = new ThreadPoolExecutor(maxSize, maxSize, 6, TimeUnit.MINUTES, new LinkedBlockingQueue());
        // 允许核心线程被回收,使线程池空闲时会收缩至0
        pool.allowCoreThreadTimeOut(true);
        // 设置任务线程的名字
        pool.setThreadFactory(new SimpleThreadFactory(threadNamePrefix));
        // 设置任务拒绝策略(若不配置任务队列大小, 其实也就不存在拒绝的情况)
        pool.setRejectedExecutionHandler(new SimpleRejectedExecutionHandler());
        return pool;
    }

    // 默认拒绝策略 ThreadPoolExecutor.AbortPolicy
    static class SimpleRejectedExecutionHandler implements RejectedExecutionHandler {
        
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            log.error("Task {} rejected from {}", r.toString(), e.toString());
        }
    }

    // 参考Executors.defaultThreadFactory()
    static class SimpleThreadFactory implements ThreadFactory {
        private final ThreadGroup group;
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private final String namePrefix;

        SimpleThreadFactory(String namePrefix) {
            SecurityManager s = System.getSecurityManager();
            group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
            this.namePrefix = namePrefix + "-";
        }

        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0);
            if (t.isDaemon()) t.setDaemon(false);
            if (t.getPriority() != Thread.NORM_PRIORITY) t.setPriority(Thread.NORM_PRIORITY);
            return t;
        }
    }
}

线程池使用模板代码

import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

// 线程池使用模板代码
public class ThreadPoolUtilTest {

    private static final Logger log = LoggerFactory.getLogger(ThreadPoolUtilTest.class);
    private static final String poolName = "test";
    private static final ThreadPoolExecutor pool = ThreadPoolUtil.buildPool(poolName, 4);

    // 不阻塞调用线程
    @Test
    public void noBlockingExecute() {
        List taskList = createList();// 待处理数据

        for (Object task : taskList) {
            pool.execute(() -> {
                try {
        服务器托管网            // 数据处理业务逻辑
                } catch (Exception e) {
                    log.error("task error: {}", task.toString());
                    log.error("exception: ", e);
                }
            });
        }
        // 线程池不再接受新任务,当前任务队列处理完后关闭线程池, 此方法不会阻塞调用线程
        pool.shutdown();
    }

    // 阻塞调用线程, 每次任务线程跑完计算进度
    @Test
    public void blockingExecute() {
        List taskList = createList();// 待处理数据
        int taskNum = taskList.size();
        int oneOfTenSize = (taskNum  {
                try {
                    // 数据处理业务逻辑
                    log.info("start {}", task);
                    TimeUnit.MILLISECONDS.sleep(3);
                    log.info("end {}", task);
                    //
                } catch (Exception e) {
                    log.error("task error: {}", task.toString());
                    log.error("exception: ", e);
              服务器托管网  } finally {
                    // 线程池已完成本批次任务的数量
                    // long completedTaskCount = pool.getCompletedTaskCount() - beginCompletedTaskCount + 1;
                    long completedTaskCount = taskNum - countDownLatch.getCount() + 1;
                    if (completedTaskCount % oneOfTenSize == 0) {
                        // 存在同一进度被多个线程打印的情况
                        log.info("进度: {}/{}", completedTaskCount, taskNum);
                    }
                    countDownLatch.countDown();
                }
            });
        }
        pool.shutdown();
        try {
            countDownLatch.await();
            log.info("本批次任务执行完毕");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    // 阻塞调用线程, 使用异步线程打印进度
    @Test
    public void blockingExecuteWithMonitorThread() {
        List taskList = createList();// 待处理数据

        int taskNum = taskList.size();
        // 任务完成进度监控
        ThreadPoolUtil.progressMonitor(pool, poolName, pool.getCompletedTaskCount(), taskNum);

        // 多线程执行任务
        CountDownLatch countDownLatch = new CountDownLatch(taskNum);
        for (Object task : taskList) {
            pool.execute(() -> {
                try {
                    // 数据处理业务逻辑
                    log.info("start {}", task);
                    TimeUnit.MILLISECONDS.sleep(3);
                    log.info("end {}", task);
                    //
                } catch (Exception e) {
                    log.error("task error: {}", task.toString());
                    log.error("exception: ", e);
                } finally {
                    countDownLatch.countDown();
                }
            });
        }
        pool.shutdown();
        try {
            countDownLatch.await();
            log.info("本批次任务执行完毕");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private List createList() {
        List taskList = new ArrayList();
        for (int i = 0; i 

cook

  • 线程池业务逻辑必须使用try-catch包裹, 否则异常发生时日志中不会有任何异常信息, 不利于异常时排查
  • ThreadPoolExecutor#getCompletedTaskCount 方法使用的ReentrantLock锁, 读和写都会加锁, 测试发现分别使用 每次任务完成后使用getCompletedTaskCount 计算进度 和 新建线程定时计算进度ThreadPoolUtil#progressMonitor 两种方式时, 整个批次完成耗时相差无几

服务器托管,北京服务器托管,服务器租用 http://www.fwqtg.net
机房租用,北京机房租用,IDC机房托管, http://www.fwqtg.net

相关推荐: Doris写入数据异常提示actual column number in csv file is less than schema column number

版本信息: Flink 1.17.1 Doris 1.2.3 Flink Doris Connector 1.4.0 写入方式 采用 String 数据流,依照社区网站的样例代码,在sink之前将数据转换为DataStream,分隔符采用”t”。 运行异常 通…

Related posts:

  1. 外贸服务器托管:提升运营效率的最佳选择
  2. 浙江优秀的服务器托管服务推荐
  3. 北京怎么办idc证
  4. 重庆易用托管服务器:高效稳定,简单易用
  5. 服务器托管公司的优点

服务器托管,北京服务器托管,服务器租用,机房机柜带宽租用

服务器托管

咨询:董先生

电话13051898268 QQ/微信93663045!

上一篇: 腾讯的认证到底有没有用?怎么样可以通过考试?
下一篇: kubernates的集群安装-kubadm

最新更新

  • input输入时的边框样式去除
  • 安卓端出现https请求失败的一次问题排查
  • Java并发编程基础
  • AntDesignBlazor示例——创建列表页
  • #yyds干货盘点# LeetCode程序员面试金典:矩阵中的最长递增路径

随机推荐

  • 服务器 CPU 与家用 CPU 有什么区别?
  • 北京电信有几个idc机房
  • Java的static修饰符
  • 湖南联通大带宽服务器租用云空间
  • 阿里云服务器投资托管方案介绍

客服咨询

  • 董先生
  • 微信/QQ:93663045
  • 电话:13051898268
  • 邮箱:dongli@hhisp.com
  • 地址:北京市石景山区重聚园甲18号2层

友情链接

  • 服务器托管
  • 机房租用托管
  • 服务器租用托管
©2023 服务器托管,北京服务器托管,服务器租用-价格及机房咨询 京ICP备13047091号-8