使用自定义FutureTask实现大小不固定的定时线程池

在使用定时任务执行线程池ScheduledThreadPoolExecutor时，在相应的定义中，只需要通过coreSize来定义1个大小固定的线程池，并且在其具体的定义中，因为队列长度是无限的，因此maxSize实际上也没有任何作用。其问题在于，如果coreSize定义过大，则会造成线程池中大量的空闲线程，实际上没有任务可作.

如定义1个coreSize为10的定时线程池，即使只周期性的执行1个任务，在一段时间之后,其池中的执行线程会逐渐增多，直到达到coreSize上限. 并且由于定时调度的原因，不能设置 allowCoreThreadTimeOut， 此设置会导致定时的任务因为无线程可用而不会触发,原因在于任务的定时是通过执行线程的takeTask操作被动触发的。

本文通过一个单线程定时线程池和一个额外的普通的ExecutorService进行协作，定时线程池只负责调度，而具体的执行则交给执行线程池来处理。而执行线程池的线程本身是可以设置或处理达到线程数可调节。通过2个线程池之间进行协作，完成调度完成。

同时，本实现也将专门处理 scheduleWithFixedDelay 或 scheduleWithFixedDelay 中的延时处理，保证任务必须在前1个任务执行完成之后才处理下一个任务，而避免简单调度中可能产生同1个任务由于执行超时出现并行执行的问题.

本文将定义定时线程池称之为 defineScheduledExecutor,类型为ScheduledThreadPoolExecutor； 实际执行线程池称之为 realExecutor，类型为 ExecutorService

继续阅读“使用自定义FutureTask实现大小不固定的定时线程池”

实现优先使用运行线程及调整线程数大小的线程池

当前在JDK中默认使用的线程池 ThreadPoolExecutor，在具体使用场景中，有以下几个缺点

  1. core线程一般不会timeOut
  2. 新任务提交时，如果工作线程数小于 coreSize，会自动先创建线程，即使当前工作线程已经空闲，这样会造成空闲线程浪费
  3. 设置的maxSize参数只有在队列满之后，才会生效，而默认情况下容器队列会很大(比如1000)

如一个coreSize为10，maxSize为100,队列长度为1000的线程池，在运行一段时间之后的效果会是以下2个效果:

  1. 系统空闲时，线程池中始终保持10个线程不变，有一部分线程在执行任务，另一部分线程一直wait中(即使设置allowCoreThreadTimeOut)
  2. 系统繁忙时，线程池中线程仍然为10个，但队列中有还没有执行的任务(不超过1000),存在任务堆积现象

本文将描述一下简单版本的线程池，参考于 Tomcat ThreadPoolExecutor, 实现以下3个目标

  1. 新任务提交时，如果有空闲线程，直接让空闲线程执行任务，而非创建新线程
  2. 如果coreSize满了，并且线程数没有超过maxSize，则优先创建线程，而不是放入队列
  3. 其它规则与ThreadPoolExecutor一致，如 timeOut机制
继续阅读“实现优先使用运行线程及调整线程数大小的线程池”

一种在线程池中透传或继承ThreadLocal信息的方法

在实际的业务代码中，经常会使用到 ThreadLocal 用于跨业务代码 来获取在上游设置的值。比如，在spring mvc中 spring web mvc 中通过 RequestContextHolder 设置 HttpServletRequest，业务代码则可以在 controller 或者是 service中 通过 RequestContextHolder#getRequestAttributes 获取相应的对象. 但这种方法有一个限制，即 setValue 和 getValue 的代码必须在同一个线程内. 当然，这也是属于通过 ThreadLocal 来避免竞争的一种手法.

针对之前已经可以工作的代码，如果将相应的业务代码 迁移至一个新的线程池中运行，即封装为 1个 runnable 对象，那么相应的代码即不能正确地工作了。

如下参考所示 

ThreadLocal<String> local = new ThreadLocal<>();
local.set("value1");

//打印 获取->value1
System.out.println("获取->" + local.get());

Runnable runnable = () -> {
    System.out.println("获取->" + local.get());
};
//打印 获取->null
new Thread(runnable).start();

上面的 sout 可以是任意1个业务上的 method 调用。仅仅是将 method调用 转由新线程来运行，相应的业务逻辑即不能正常工作。 至于这里将调用转由新线程来运行，可以有很多的场景。如 支持 timedout 调用(利用future.get(timed))。

本文通过反射调用读取当前Thread的信息，将值注入到新线程中的threadLocal中，以达到透传threadLocal的目的。不需要修改任何业务代码，也不需要使用InheritableThreadLocal(此类也并不用于当前场景)

主要的实现基于以下步骤

  • 提前提取当前线程ThreadLocal变量值
  • 执行时复制至新线程中
  • 执行结束之后删除未变化值

本文中的代码均基于反射调用，需要打开相应的 Accessible 属性.

继续阅读“一种在线程池中透传或继承ThreadLocal信息的方法”

使用FeignClient进行服务调用时的各类错误列表及快速定位

基于spring cloud 体系的微服务，服务间调用默认使用 openFeign 进行调用，但是由于源码本身的问题。针对于在调用时出现的各类信息，并没有分门别类的错误描述。当业务需要针对不同的错误进行处理时，就不能简单的通过FeignException来进行区分。笔者通过整理服务间调用时的各类异常堆栈及列表。分别列出相应的错误列表及触发时堆栈，以方便后续分别进行业务判断和处理

本应用基于ribbon进行服务查找,使用spring-cloud open-feign, 没有使用retry机制, 底层使用feign-okhttp, 因此，如果有异常与下列堆栈不一致时，请检查是否一致. 

版本: openFeign: 9.7.0, netflix ribbon: 2.2.5,  okhttp3: 3.8.1

总共的异常列表如下所示

  • 无服务时异常
  • 有服务但连接失败
  • 调用中网络错误读取数据超时
  • 响应4xx错误码
  • 响应5xx错误码
  • 解析结果错误(json反序列化失败)
  • 解析结果错误(无converter)

所有的异常触发类均由类 SynchronousMethodHandler 触发，但作用行不一样.
以下异常栈从原始cause，依次往下。如 cause1为 root cause, 最下层为业务层接收到的异常

继续阅读“使用FeignClient进行服务调用时的各类错误列表及快速定位”

在基于spring体系的业务中正确地关闭线程池

在业务代码中，特别是基于spring体系的代码中，均会使用线程池进行一些操作，比如异步处理消息，定时任务，以及一些需要与当前业务分离开的操作等。常规情况下，使用spring体系的TaskExecutor或者是自己定义ExecutorService，均可以正常地完成相应的操作。不论是定义一个spring bean，或者是使用 static Thread工具类均是能满足条件。

但是，如果需要正常地关闭spring容器时，这些线程池就不一定能够按照预期地关闭了。结果就是，当使用代码 context.close() 时，期望进程会正常地退出，但实际上进程并不会退出掉.原因就在于这些线程池中还在运行的线程。
本文描述了在基于spring boot的项目中，如何正确地配置线程池，以保证线程池能够正确的在整个spring容器周期内运行，并且在容器正常关闭时能够一并退出掉.

  • 定义bean时添加destroyMethod方法或相应生命周期方法
  • 设置线程池中线程为daemon
  • 为每个线程池正确地命名及使用ThreadFactory
  • 丢弃不再需要的周期性任务
  • 监听ContextClosedEvent，触发额外操作
继续阅读“在基于spring体系的业务中正确地关闭线程池”

spring cloud consul微服务长轮询更新

在使用spring cloud体系进行微服务开发时，一般的服务注册和发现除了 eureka 之外，还可以采用consul, zookeeper来实现。本文描述使用组件 spring-cloud-consul 时，如果保证服务变更时的实时发现和处理.

一般来说，采用zookeeper这种服务端通过连接挂在临时节点，然后消费端通过监听节点来感知服务的变更，这种实现方式对于服务端的上下线感知是最快的。因为通过socket来维系服务的健康度，当服务不可用时，socket立马即会断开.(本文不考虑服务假死,socket仍可用的情况),而监听方马上通过变更来更新server list. 当下一个请求处理时，即会正确选择合适的服务方进行处理。而且当服务方上线时，也会立马进行感知加入到服务列表中。

对于consul来实现服务注册，有很多种方式，这里我们采用 TTL check,即设置 spring.cloud.consul.discovery.heartbeat.enabled=true的情况,在这种情况下 其 默认ttl值为30，默认汇报时间为18左右.

对于消费者，即discover方，目前spring cloud consul中，采用标准的 ribbon-loadbalancer 中的 ServerList.getUpdatedListOfServers 来完成实例的更新。相应的定时任务通过 ServerListUpdater.start (默认实现为 PollingServerListUpdater) 来开启，内部采用 scheduleWithFixedDelay(默认30秒周期) 来实现.

在 consul 层,实现方为 ConsulServerList#getServers,即通过调用ConsulClient#getHealthServices 来获取健康的服务列表，对应consul的api即 /health/service/:service 来完成.

在这种情况下，存在一个间隔期，即30秒周期。如果服务方变更了，discover并不能实时感知，还需要等待下一次轮询才能感知到，对于微服务来说，这即是一个服务不可用的场景.

在官方 consul 的 API中，是存在 阻塞式调用的，即响应并不会立即返回，直到有变化或wait时间到。即长轮询概念，可以采取类似长轮询的概念来达到数据第一时间感知。目前世面大部分的配置中心同样采取类似的技术。

继续阅读“spring cloud consul微服务长轮询更新”