共计 6268 个字符,预计需要花费 16 分钟才能阅读完成。
原创:扣钉日记(微信公众号 ID:codelogs),欢送分享,转载请保留出处。
简介
最近我察看到一个景象,当服务的申请量突发的增长一下时,服务的无效 QPS 会降落很多,有时甚至会降到 0,这种景象网上也偶有提到,但少有解释得分明的,所以这里来分享一下问题成因及解决方案。
队列提早
目前的 Web 服务器,如 Tomcat,申请处理过程大略都相似如下:
这是 Tomcat 申请解决的过程,如下:
- Acceptor 线程:线程名相似 http-nio-8080-Acceptor-0,此线程用于接管新的 TCP 连贯,并将 TCP 连贯注册到 NIO 事件中。
- Poller 线程:线程名相似 http-nio-8080-ClientPoller-0,此线程个别有 CPU 核数个,用于轮询已连贯的 Socket,接管新到来的 Socket 事件(如调用端发申请数据了),并将沉闷 Socket 放入 exec 线程池的申请队列中。
- exec 线程:线程名相似 http-nio-8080-exec-0,此线程从申请队列中取出沉闷 Socket,并读出申请数据,最初执行申请的 API 逻辑。
这里不必太关怀 Acceptor
与Poller
线程,这是 nio 编程时常见的线程模型,咱们将重点放在 exec 线程池上,尽管 Tomcat 做了一些优化,但它还是从 Java 原生线程池扩大进去的,即有一个工作队列与一组线程。
当申请量突发增长时,会产生如下的状况:
- 当申请量不大时,工作队列根本是空的,每个申请都能失去及时的解决。
- 但当申请量突发时,工作队列中就会有很多申请,这时排在队列前面的申请,就会被解决得越晚,因此申请的整体耗时就会变长,甚至十分长。
可是,exec 线程们还是在一刻不停歇的解决着申请的呀,按理说服务 QPS 是不会缩小的呀!
简略想想的确如此,但调用端个别是有超时工夫设置的,不会有限期待上来,当客户端期待超时的时候,这个申请实际上 Tomcat 就不必再解决了,因为就算解决了,客户端也不会再去读响应数据的。
因而,当队列比拟长时,队列前面的申请,基本上都是不必再解决的,但 exec 线程池不晓得啊,它还是会判若两人地解决这些申请。
当 exec 线程执行这些已超时的申请时,若又有新申请进来,它们也会排在队尾,这导致这些新申请也会超时,所以在流量突发的这段时间内,申请的无效 QPS 会降落很多,甚至会降到 0。
这种超时也叫做队列提早,但队列在软件系统中利用得太宽泛了,比方操作系统调度器保护了线程队列,TCP 中有 backlog 连贯队列,锁中保护了期待队列等等。
因而,很多零碎也会存在这种景象,平时响应工夫挺稳固的,但偶然耗时很高,这种状况有很多都是队列提早导致的。
优化队列提早
晓得了问题产生的起因,要优化它就比较简单了,咱们只须要让队列中那些长时间未解决的申请临时让路,让线程去执行那些等待时间不长的申请即可,毕竟这些长时间未解决的申请,让它们再等等也无防,因为客户端可能曾经超时了而不须要申请后果了,尽管这毁坏了队列的公平性,但这是咱们须要的。
对于 Tomcat,在 springboot 中,咱们能够如下批改:
应用 WebServerFactoryCustomizer 自定义 Tomcat 的线程池,如下:
@Component
public class TomcatExecutorCustomizer implements WebServerFactoryCustomizer<TomcatServletWebServerFactory> {
@Resource
ServerProperties serverProperties;
@Override
public void customize(TomcatServletWebServerFactory factory) {
TomcatConnectorCustomizer tomcatConnectorCustomizer = connector -> {ServerProperties.Tomcat.Threads threads = serverProperties.getTomcat().getThreads();
TaskQueue taskqueue = new SlowDelayTaskQueue(1000);
ThreadPoolExecutor executor = new org.apache.tomcat.util.threads.ThreadPoolExecutor(threads.getMinSpare(), threads.getMax(), 60L, TimeUnit.SECONDS,
taskqueue, new CustomizableThreadFactory("http-nio-8080-exec-"));
taskqueue.setParent(executor);
ProtocolHandler handler = connector.getProtocolHandler();
if (handler instanceof AbstractProtocol) {AbstractProtocol<?> protocol = (AbstractProtocol<?>) handler;
protocol.setExecutor(executor);
}
};
factory.addConnectorCustomizers(tomcatConnectorCustomizer);
}
}
留神,这里还是应用的 Tomcat 实现的线程池,只是将工作队列 TaskQueue 扩大为了 SlowDelayTaskQueue,它的作用是将长时间未解决的工作移到另一个慢队列中,待以后队列中无工作时,再把慢队列中的工作移回来。
为了能记录工作入队列的工夫,先封装了一个记录时间的工作类 RecordTimeTask,如下:
@Getter
public class RecordTimeTask implements Runnable {
private Runnable run;
private long createTime;
private long putQueueTime;
public RecordTimeTask(Runnable run){
this.run = run;
this.createTime = System.currentTimeMillis();
this.putQueueTime = this.createTime;
}
@Override
public void run() {run.run();
}
public void resetPutQueueTime() {this.putQueueTime = System.currentTimeMillis();
}
public long getPutQueueTime() {return this.putQueueTime;}
}
而后队列的扩大实现如下:
public class SlowDelayTaskQueue extends TaskQueue {
private long timeout;
private BlockingQueue<RecordTimeTask> slowQueue;
public SlowDelayTaskQueue(long timeout) {
this.timeout = timeout;
this.slowQueue = new LinkedBlockingQueue<>();}
@Override
public boolean offer(Runnable o) {
// 将工作包装一下,目标是为了记录工作放入队列的工夫
if (o instanceof RecordTimeTask) {return super.offer(o);
} else {return super.offer(new RecordTimeTask(o));
}
}
public void pullbackIfEmpty() {
// 如果队列空了,从慢队列中取回来一个
if (this.isEmpty()) {RecordTimeTask r = slowQueue.poll();
if (r == null) {return;}
r.resetPutQueueTime();
this.add(r);
}
}
@Override
public Runnable poll(long timeout, TimeUnit unit) throws InterruptedException {pullbackIfEmpty();
while (true) {RecordTimeTask task = (RecordTimeTask) super.poll(timeout, unit);
if (task == null) {return null;}
// 申请在队列中长时间期待,移入慢队列中
if (System.currentTimeMillis() - task.getPutQueueTime() > this.timeout) {this.slowQueue.offer(task);
continue;
}
return task;
}
}
@Override
public Runnable take() throws InterruptedException {pullbackIfEmpty();
while (true) {RecordTimeTask task = (RecordTimeTask) super.take();
// 申请在队列中长时间期待,移入慢队列中
if (System.currentTimeMillis() - task.getPutQueueTime() > this.timeout) {this.slowQueue.offer(task);
continue;
}
return task;
}
}
}
逻辑其实挺简略的,如下:
- 当工作入队列时,包装一下工作,记录一下入队列的工夫。
- 而后线程从队列中取出工作时,若发现工作等待时间过长,就将其移入慢队列。
- 而 pullbackIfEmpty 的逻辑,就是当队列为空时,再将慢队列中的工作移回来执行。
为了将申请的队列提早记录在 access.log 中,我又批改了一下 Task,并加了一个 Filter,如下:
-
应用 ThreadLocal 将队列提早先存起来
@Getter public class RecordTimeTask implements Runnable {private static final ThreadLocal<Long> WAIT_IN_QUEUE_TIME = new ThreadLocal<>(); private Runnable run; private long createTime; private long putQueueTime; public RecordTimeTask(Runnable run){ this.run = run; this.createTime = System.currentTimeMillis(); this.putQueueTime = this.createTime; } @Override public void run() { try {WAIT_IN_QUEUE_TIME.set(System.currentTimeMillis() - this.createTime); run.run();} finally {WAIT_IN_QUEUE_TIME.remove(); } } public void resetPutQueueTime() {this.putQueueTime = System.currentTimeMillis(); } public long getPutQueueTime() {return this.putQueueTime;} public static long getWaitInQueueTime(){return ObjectUtils.defaultIfNull(WAIT_IN_QUEUE_TIME.get(), 0L); } }
-
再在 Filter 中将队列提早取出来,放入 Request 对象中
@WebFilter @Component public class WaitInQueueTimeFilter extends HttpFilter { @Override public void doFilter(HttpServletRequest request, HttpServletResponse response, FilterChain chain) throws IOException, ServletException {long waitInQueueTime = RecordTimeTask.getWaitInQueueTime(); // 将等待时间设置到 request 的 attribute 中,给 access.log 应用 request.setAttribute("waitInQueueTime", waitInQueueTime); // 如果申请在队列中期待了太长时间,客户端大概率已超时,就没有必要再执行了 if (waitInQueueTime > 5000) {response.sendError(503, "service is busy"); return; } chain.doFilter(request, response); } }
-
而后在 access.log 中配置队列提早
server: tomcat: accesslog: enabled: true directory: /home/work/logs/applogs/java-demo file-date-format: .yyyy-MM-dd pattern: '%h %l %u %t"%r"%s %b %Dms %{waitInQueueTime}rms"%{Referer}i""%{User-Agent}i" "%{X-Forwarded-For}i"'
留神,在 access.log 中配置
%{xxx}r
示意取申请 xxx 属性的值,所以,%{waitInQueueTime}r
就是队列提早,前面的 ms 是毫秒单位。
优化成果
我应用接口压测工具 wrk 压了一个测试接口,此接口执行工夫 100ms,应用 1000 个并发去压,1s 的超时工夫,如下:
wrk -d 10d -T1s --latency http://localhost:8080/sleep -c 1000
而后,用 arthas 看一下线程池的队列长度,如下:
[arthas@619]$ vmtool --action getInstances \
--classLoaderClass org.springframework.boot.loader.LaunchedURLClassLoader \
--className org.apache.tomcat.util.threads.ThreadPoolExecutor \
--express 'instances.{#{"ActiveCount":getActiveCount(),"CorePoolSize":getCorePoolSize(),"MaximumPoolSize":getMaximumPoolSize(),"QueueSize":getQueue().size()} }' \
-x 2
能够看到,队列长度远小于 1000,这阐明队列中积压得不多。
再看看 access.log,如下:
能够发现,尽管队列提早任然存在,但被管制在了 1s 以内,这样这些申请就不会超时了,Tomcat 的无效 QPS 保住了。
而最初面那些队列提早极长的申请,则是被不偏心看待的申请,但只能这么做,因为在申请量超出 Tomcat 解决能力时,只能就义掉它们,以顾全大局。