为什么用线程池

有时候，系统需要处理非常多的执行时间很短的请求，如果每一个请求都开启一个新线程的话，系统就要不断的进行线程的创建和销毁，有时花在创建和销毁线程上的时间会比线程真正执行的时间还长。而且当线程数量太多时，系统不一定能受得了

使用线程池主要为了解决一下几个问题：

降低资源消耗：通过重用线程池中的线程，来减少每个线程创建和销毁的性能开销

提高响应速度：当任务到达时，任务可以不需要等到线程创建就能立即执行

提高线程的可管理性：线程是稀缺资源，如果无限制地创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一维护和管理，分配、调优和监控

线程池规则

线程池的线程执行规则跟任务队列有很大的关系。

下面都假设任务队列没有大小限制：

如果线程数量 <= 核心线程数量，那么 直接启动一个核心线程 来执行任务，不会放入队列中。

如果线程数量 > 核心线程数，但 <= 最大线程数，并且任务队列是LinkedBlockingDeque的时候，超过核心线程数量的任务会放在任务队列中排队。
如果线程数量 > 核心线程数，但 <= 最大线程数，并且任务队列是SynchronousQueue的时候，线程池会创建新线程执行任务，这些任务也不会被放在任务队列中。这些线程属于非核心线程，在任务完成后，闲置时间达到了超时时间就会被清除。

如果线程数量 > 核心线程数，并且 > 最大线程数，当任务队列是LinkedBlockingDeque，会将超过核心线程的任务放在任务队列中排队。也就是当任务队列是LinkedBlockingDeque并且没有大小限制时，线程池的最大线程数设置是无效的，他的线程数最多不会超过核心线程数。
如果线程数量 > 核心线程数，并且 > 最大线程数，当任务队列是SynchronousQueue的时候，会因为线程池 拒绝添加 任务而抛出异常。 

任务队列大小有限时

当LinkedBlockingDeque塞满时，新增的任务会直接创建新线程来执行，当创建的线程数量超过最大线程数量时会抛异常。
SynchronousQueue没有数量限制。因为他根本不保持这些任务，而是直接交给线程池去执行。当任务数量超过最大线程数时会直接抛异常。

常用类

Executor

Executor是一个接口，跟线程池有关的基本都要跟他打交道。下面是常用的ThreadPoolExecutor的关系

Executor接口很简单，只有一个execute方法

ExecutorService是Executor的子接口，增加了一些常用的对线程的控制方法，之后使用线程池主要也是使用这些方法

AbstractExecutorService是一个抽象类。ThreadPoolExecutor就是实现了这个类。

ThreadPoolExecutor

ThreadPoolExecutor的使用

线程池的创建

构造方法

构造方法参数说明

• corePoolSize
核心线程数，默认情况下核心线程会一直存活，即使处于闲置状态也不会受存keepAliveTime限制。除非将allowCoreThreadTimeOut设置为true。

  当提交一个任务到线程池时，线程池会创建一个线程来执行任务，即使其他空闲的基本线程能够执行新任务也会创建线程，等到需要执行的任务数大于线程池基本大小时就不再创建。如果调用了prestartAllCoreThreads()方法，线程池会提前创建并启动所有基本线程

• maximumPoolSize
  线程池所能容纳的最大线程数。超过这个数的线程将被阻塞。当任务队列为没有设置大小的LinkedBlockingDeque时，这个值无效。
  如果队列满了，并且已创建的线程数小于最大线程数，则线程池会再创建新的线程执行任务。值得注意的是，如果使用了无界的任务队列这个参数无效

• keepAliveTime
非核心线程的闲置超时时间，工作线程空闲后保持存活的时间，超过这个时间就会被回收，如果任务很多，并且每个任务执行的时间比较短，可以调大时间，提高线程利用率

• unit
  指定keepAliveTime的单位，如TimeUnit.SECONDS。当将allowCoreThreadTimeOut设置为true时对corePoolSize生效

  天（Days）、小时（HOURS）、分钟（MINUTES）、毫秒（MILLISECONDS）、微秒（MICROSECONDS，千分之一毫秒）和纳秒（NANOSECONDS，千分之一微秒）
  

• workQueue
  线程池中常用的三种任务队列，SynchronousQueue , LinkedBlockingDeque, ArrayBlockingQueue

  1 ArrayBlockingQueue: 是一个基于数组结构的有界阻塞队列，按FIFO原则进行排序

  2 LinkedBlockingQueue: 一个基于链表结构的阻塞队列，吞吐量高于ArrayBlockingQueue。静态工厂方法Excutors.newFixedThreadPool()使用了这个队列

  3 SynchronousQueue: 一个不存储元素的阻塞队列。每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态，吞吐量高于LinkedBlockingQueue，静态工厂方法Excutors.newCachedThreadPool()使用了这个队列

  4 PriorityBlockingQueue: 一个具有优先级的无限阻塞队列

• threadFactory
线程工厂，提供创建新线程的功能。ThreadFactory是一个接口，只有一个方法

  通过线程工厂可以对线程的一些属性进行定制，比如为每个创建出来的线程设置更有意义的名字
  
  
  public interface ThreadFactory {
    Thread newThread(Runnable r);
  }
  

• RejectedExecutionHandler
  RejectedExecutionHandler也是一个接口，只有一个方法，当线程池中的队列和线程池已经全部使用，说明线程池处于饱和状态，添加新线程被拒绝时，会调用RejectedExecutionHandler的rejectedExecution方法

  AbortPolicy : 直接抛出异常，默认情况下采用这种策略
  CallerRunsPolicy :只用调用者所在线程来运行任务
  DiscardOldestPolicy : 丢弃队列里最近的一个任务，并执行当前任务
  DiscardPolicy : 不处理，丢弃掉
  
  更多的时候，我们应该通过实现RejectedExecutionHandler 接口来自定义策略，比如记录日志或持久化存储等
  

  public interface RejectedExecutionHandler {
    void rejectedExecution(Runnable var1, ThreadPoolExecutor var2);
  }
  

提交任务

可以使用execute和submit两个方法向线程池提交任务

execute方法用于提交不需要返回值的任务，利用这种方式提交的任务无法得知是否正常执行

threadPoolExecutor.execute(new Runnable() {
			
			@Override
			public void run() {
				try {
					Thread.sleep(5000);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		});

submit方法用于提交一个任务并带有返回值，这个方法将返回一个Future类型对象。可以通过 这个返回对象判断任务是否执行成功，并且可以通过future.get()方法来获取返回值，get()方法会阻塞当前线程直到任务完成

Future<?> future = threadPoolExecutor.submit(futureTask);
Object value = future.get();

关闭线程池

可以通过调用线程池的shutdown或shutdownNow方法来关闭线程池。他们的原理是遍历线程池中的工作线程，然后逐个调用线程的interrupt方法来中断线程，所以无响应中断的任务可能永远无法停止。

但是他们存在一定的区别

shutdownNow 首先 将线程池的状态 设置为STOP，然后 尝试停止 所有 正在执行或暂停任务 的线程，并 返回等待执行任务的列表
shutdown 只是 将线程池的状态 设置成 SHUTDOWN 状态，然后 中断 所有正在执行的任务

只要调用了这两个关闭方法的一个，isShutdown就会返回true。当所有的任务都关闭后，才表示线程池关闭成功，这时调用isTerminated方法会返回true

至于应该调用哪一种方法来关闭线程池，应该由提交到线程池的任务特性决定，通常调用shutdown方法来关闭线程池，如果任务不一定执行完，则可以调用shutdownNow方法。

合理配置线程池

要想合理地配置线程池，首先要分析任务特性

任务的性质：CPU密集型任务、IO密集型任务和混合型任务。
任务的优先级：高、中和低。
任务的执行时间：长、中和短。
任务的依赖性：是否依赖其他系统资源，如数据库连接。

性质不同的任务可以用不同规模的线程池分开处理。

CPU密集型任务应该配置尽可能少的线程，如配置N+1个线程，N位CPU的个数。

IO密集型任务线程并不是一直在执行任务，则应配置尽可能多的线程，如2*N。

混合型任务，如果可以拆分，将其拆分成一个CPU密集型任务和一个IO密集型任务，只要这两个任务执行的时间相差不是太大，那么分解后执行的吞吐量将高于串行执行的吞吐量

优先级不同的任务可以交给优先级队列PriorityBlcokingQueue来处理。

执行时间不同的任务可以交给不同规模的线程池来处理。

依赖数据库的任务，因此线程提交SQL后需要等待数据库返回结果，等待的时间越长，则CPU空闲时间越长，那么线程数应该设置的越大，这样能更好滴利用CPU。

应用示例

验证shutdown和shutdownNow的区别

首先构造一个线程池，用ArrayBlockingQueue作为其等待队列，队列初始化容量为10。该线程池核心容量为 10，最大容量为20，线程存活时间为1分钟

static BlockingQueue blockingQueue=new ArrayBlockingQueue<>(10);
static ThreadPoolExecutor threadPoolExecutor=new ThreadPoolExecutor(10, 20, 1, TimeUnit.MINUTES, blockingQueue);

构造一个实现Runable接口的类TaskWithoutResult，其逻辑很简单，睡眠1秒

/**
 * 无返回值的任务
 */
class TaskWithoutResult implements Runnable {
	private int sleepTime=1000;//默认睡眠时间1s
	public TaskWithoutResult(int sleepTime) {
		this.sleepTime=sleepTime;
	}
	@Override
	public void run() {
		System.out.println("线程"+Thread.currentThread()+"开始运行");
		try {
			Thread.sleep(sleepTime);
		} catch (InterruptedException e) {//捕捉中断异常
			System.out.println("线程"+Thread.currentThread()+"被中断");
		}
		System.out.println("线程"+Thread.currentThread()+"结束运行");
	}
}

/**
 * 中断测试
 */
public static void  test1() {
	for(int i=0;i<10;i++) {
		Runnable runnable=new TaskWithoutResult(1000);
		threadPoolExecutor.submit(runnable);
	}
	//threadPoolExecutor.shutdown();//不会触发中断
	threadPoolExecutor.shutdownNow();//会触发中断
}

分别测试shutdown和shutdownNow()方法，结果shutdown()方法的调用并不会引发中断，而shutdownNow()方法则会引发中断。

这也正验证前面所说的，shutdown方法只是发出了停止信号，等所有线程执行完毕会关闭线程池；而shutdownNow则是立即停止所有任务。

异步线程池ThreadPoolTaskExecutor

1 Spring是通过任务执行器(TaskExecutor)来实现多线程和并发编程
2 使用ThreadPoolTaskExecutor来创建一个基于线城池的TaskExecutor
3 实际开发任务大多数情况下都是异步非阻塞的。我们配置注解@EnableAsync可以开启异步任务。然后在实际执行的方法上配置注解@Async上声明是异步任务。

同步交互：指发送一个请求, 需要等待返回, 然后才能够发送下一个请求，有个等待过程；

异步交互：指发送一个请求, 不需要等待返回, 随时可以再发送下一个请求，即不需要等待。

区别：一个需要等待，一个不需要等待。在部分情况下，我们的项目开发中都会优先选择不需要等待的异步交互方式。

ThredPoolTaskExcutor的处理流程

 当池子大小小于corePoolSize，就新建线程，并处理请求
 当池子大小等于corePoolSize，把请求放入workQueue中，池子里的空闲线程就去workQueue中取任务并处理
 当workQueue放不下任务时，就新建线程入池，并处理请求，如果池子大小撑到了maximumPoolSize，就用RejectedExecutionHandler来做拒绝处理
 当池子的线程数大于corePoolSize时，多余的线程会等待keepAliveTime长时间，如果无请求可处理就自行销毁

引入 Maven 依赖

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
        <version>2.4.4</version>
    </dependency>
</dependencies>

异步执行的配置类 AsyncConfig

/**
 * @classname AsyncConfig
 * @description 开启异步执行的配置类
 */
@Configuration
@EnableAsync //开启异步执行
@ComponentScan("com.melodyjerry.thread")
public class AsyncConfig implements AsyncConfigurer {
    @Override
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
        //线程池中的线程的名称前缀
        threadPoolTaskExecutor.setThreadNamePrefix("SpringBoot线程池的前缀-");
        //线程池的核心线程数大小
        threadPoolTaskExecutor.setCorePoolSize(4);
        //线程池的最大线程数
        threadPoolTaskExecutor.setMaxPoolSize(8);
        //等待队列的大小
        threadPoolTaskExecutor.setQueueCapacity(25);
        //执行初始化
        threadPoolTaskExecutor.initialize();
        return threadPoolTaskExecutor;
    }
}

异步任务的执行类

/**
 * @classname AsyncTaskService
 * @description 异步任务的执行类
 */
@Service
public class AsyncTaskService {
    @Async //异步方法 不加的话就是同步方法
    public void executeAsyncTask(Integer i) {
        System.out.println(Thread.currentThread().getName() + ": "+i);
    }

    @Async //异步方法
    //@Async("getAsyncExecutor")
    public void executeAsyncTaskPlus(Integer i) {
        System.out.println(Thread.currentThread().getName() + "+1： " + (i+1));
    }
}

@Async注解表明该方法是个异步方法。

从Async注解接口可以看到，Target即可以在方法也可以在类型上，如果注解在类型上，表明该类所有的方法都是异步方法。

@Async("getAsyncExecutor")注解，它是刚刚我们在线程池配置类的里的那个配置方法的名字 getAsyncExecutor() ，加上这个后每次执行这个方法都会开启一个线程放入线程池中

测试 TestThreadApplication

/**
 * @classname TestThreadApplication
 * @description 测试异步任务
 */
@SpringBootApplication
public class TestThreadApplication {
    public static void main(String[] args) {
        ConfigurableApplicationContext context = SpringApplication.run(TestThreadApplication.class, args);

        AsyncTaskService asyncTaskService = context.getBean(AsyncTaskService.class);
        for (int i = 0; i < 10; i++) {
            asyncTaskService.executeAsyncTask(i);
            asyncTaskService.executeAsyncTaskPlus(i);
        }
        System.out.println("This Program has Begun successfully");

    }
}

或者

@RunWith (SpringRunner.class)
@SpringBootTest
public class SpringbootLearnApplicationTests {
    @Autowired
    private AsyncTaskService asyncTaskService;

    @Test
    public void contextLoads() {
    }

    @Test
    public void threadTest() {
        for (int i = 0; i < 20; i++) {
            asyncTaskService.executeAsyncTask(i);
        }
    }
}

异步

同步

腾讯面试居然跟我扯了半小时的CountDownLatch

CountDownLatch

应用

异步线程池ThreadPoolTaskExecutor进行并发处理批量操作

案例：用户在商品列表进行检索，结果集大约有100W商品，点击批量上架/下架。

springboot配置线程池使用多线程插入数据

文本里面有很多行url地址，需要的字段都包含在这些url中。最开始是使用的正常的普通方式去写入，但是量太大了，所以就尝试使用多线程来写入

springboot利用ThreadPoolTaskExecutor多线程批量插入百万级数据

ThreadPoolTaskExecutor多线程批量插入百万级数据

Springboot 在多线程下，处理事务；

   目前很多开发的小伙伴们，都开始采用Springboot了，因为给我们带来了许多的开发便利，只需要我们关注编写逻辑代码。在工作中是会有很多小伙伴在项目逐步实用到多线程、线程池等相关技术，但是在使用多线程会出各种各样的问题；事务就是其中一种相对麻烦的事情；

大家都知道Springboot  只需要使用注解标签@Transactional 就可以使用事务了；但是往往在多线程下该处理事务的办法就失效，毫无效果；那又该如何呢？
首先大家都知道控制事务有两种方式；第一种就是刚才所说的注解式事务、第二种就是编程式事务；既然第一种失效那么我们就使用第二种方式来实现嘛；废话不多说，上代码！

package com.slsp.cloud.config;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.slsp.cloud.bean.User;
import com.slsp.cloud.service.UserService;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.math.RandomUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.util.Date;
import java.util.HashSet;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 项目初始化加载
 */
@Component
@Slf4j
public class TradeApplicationRunner implements ApplicationRunner
{


    @Autowired
    private TransactionTemplate transactionTemplate;

    @Autowired
    private UserService userService;

    /**
     * 模拟多线程下事务处理
     * @param args
     * @throws Exception
     */
    @Override
    public void run(ApplicationArguments args) throws Exception
    {
        //线程池
        ExecutorService cachedThreadPool = Executors.newFixedThreadPool(5);
        for(int i=0;i<1;i++)
        {
            cachedThreadPool.execute(new Runnable()
            {
                @SneakyThrows
                public void run()
                {
                    log.info("多线程执行；");
                    //开启编程式事务；
                    transactionTemplate.execute(new TransactionCallback<Boolean>()
                    {
                        @Override
                        public Boolean doInTransaction(TransactionStatus transactionStatus)
                        {
                            try
                            {

                                User usr = new User();
                                //随机名字
                                String name = UUID.randomUUID().toString().replace("-", "");
                                System.out.println(name);
                                usr.setUserName(name);
                                usr.setSex(1);
                                //插入数据库
                                userService.save(usr);

                                //模拟报错；
                                int i=10/0;

                                String name2 = UUID.randomUUID().toString().replace("-", "");
                                System.out.println(name2);
                                User usr2 = new User();
                                usr2.setUserName(name2);
                                usr2.setSex(1);
                                userService.save(usr2);

                                return true;
                            }
                            catch (Exception e)
                            {
                                //事务回滚；
                                transactionStatus.setRollbackOnly();
                                e.printStackTrace();
                                return false;
                            }
                        }

                    });

                }
            });
        }
    }
}

打完收工；简单方便；

https://www.cnblogs.com/a5513633/p/13969222.html

/**
  2  * 带回滚的异步任务回调
  3  * 基类
  4  * @author Administrator
  5  *
  6  */
  7 public abstract class BaseCallBack implements Callable<String> {
  8 
  9     private static Logger logger = LoggerFactory.getLogger(BaseCallBack.class);
 10     /**
 11      * 需要回滚计数器
 12      */
 13     protected CountDownLatch rollBackLatch;
 14     /**
 15      * 主线程等待计数器
 16      */
 17     protected CountDownLatch mainThreadLatch;
 18     /**
 19      * 是否需要回滚
 20      */
 21     protected AtomicBoolean rollbackFlag;
 22     /**
 23      * 事务
 24      */
 25     protected PlatformTransactionManager transactionManager;
 26 
 27     protected abstract void doWork();
 28 
 29     @Override
 30     public String call() throws Exception {
 31         if (rollbackFlag.get()) {
 32             logger.info("需要回滚，直接不用执行了");
 33             mainThreadLatch.countDown();
 34             return Constant.ERROR_STR; // 如果其他线程已经报错 就停止线程
 35         }
 36         // 设置一个事务
 37         DefaultTransactionDefinition def = new DefaultTransactionDefinition();
 38         def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW); // 事物隔离级别，开启新事务，这样会比较安全些。
 39         TransactionStatus status = transactionManager.getTransaction(def); // 获得事务状态
 40         try {
 41             logger.info("业务开始处理：{}", this.getClass().getName());
 42             this.doWork();
 43             logger.info("业务处理结束：{}", this.getClass().getName());
 44             // 业务处理结束
 45             mainThreadLatch.countDown();
 46             logger.info("线程内正常 mainThreadLatch.countDown();");
 47             rollBackLatch.await();// 线程等待
 48             if (rollbackFlag.get()) {
 49                 logger.info("回滚事务：{}", this.getClass().getName());
 50                 transactionManager.rollback(status);
 51             } else {
 52                 logger.info("提交事务：{}", this.getClass().getName());
 53                 transactionManager.commit(status);
 54             }
 55             return Constant.SAVE_SUCCESS;
 56         } catch (Exception e) {
 57             e.printStackTrace();
 58             // 如果出错了 就放开锁 让别的线程进入提交/回滚 本线程进行回滚
 59             rollbackFlag.set(true);
 60             transactionManager.rollback(status);
 61             rollBackLatch.countDown();
 62             mainThreadLatch.countDown();
 63             logger.info("线程内异常 mainThreadLatch.countDown();");
 64             return "操作失败：" + e.getMessage();
 65         }
 66     }
 67 
 68     public CountDownLatch getRollBackLatch() {
 69         return rollBackLatch;
 70     }
 71 
 72     public void setRollBackLatch(CountDownLatch rollBackLatch) {
 73         this.rollBackLatch = rollBackLatch;
 74     }
 75 
 76     public CountDownLatch getMainThreadLatch() {
 77         return mainThreadLatch;
 78     }
 79 
 80     public void setMainThreadLatch(CountDownLatch mainThreadLatch) {
 81         this.mainThreadLatch = mainThreadLatch;
 82     }
 83 
 84     public AtomicBoolean getRollbackFlag() {
 85         return rollbackFlag;
 86     }
 87 
 88     public void setRollbackFlag(AtomicBoolean rollbackFlag) {
 89         this.rollbackFlag = rollbackFlag;
 90     }
 91 
 92     public PlatformTransactionManager getTransactionManager() {
 93         return transactionManager;
 94     }
 95 
 96     public void setTransactionManager(PlatformTransactionManager transactionManager) {
 97         this.transactionManager = transactionManager;
 98     }
 99 
100 }
复制代码