前兩天vos線上服務遇到一個問題,定時任務突然全部停下來不跑了。看日誌也沒發現什麼明顯的異常輸出,加上比較忙，所以乾脆直接使用重啟大法。沒想到過了兩天突然發現定時任務又全部停了,恰好正在趕需求,vos又是一個公司內部使用的系統，優先順序較低，索性繼續重啟等有空了再來看。然後過了1天多定時任務㕛掛了。只能暫時放下手頭上的事看一下到底是什麼導致的這個問題。

由於日誌未輸出任何異常資訊,所以要解決該問題只能靠分析程式碼了。由於vos系統使用的定時任務框架是spring schduler，所以打算了解一下spring schduler的執行原理。

一、spring schduler 的執行原理

一般我們要使用schuduler,都會往spring的配置檔案中加入下面這行。

<task:annotation-driven />

然後通過xml或者註解@Scheduled的方式配置具體的定時任務。下面我們圍繞註解的方式講解一下spring schduler是怎麼解析和執行配置的定時任務的。

1. 配置解析

spirng在解析配置檔案的時候遇到的時候,會往spring容器中注入 ScheduledAnnotationBeanPostProcessor 物件例項。具體的解析過程這裡就不詳述了。感興趣的可以自行百度或者看一下下面這個連結。
http://blog.csdn.net/tt50335971/article/details/52055755

2. ScheduledAnnotationBeanPostProcessor類

ScheduledAnnotationBeanPostProcessor 是BeanPostProcessor的子類。在spring bean的生命週期中,所有的bean構造完後都會執行postProcessAfterInitialization()方法。因此，spring容器注入ScheduledAnnotationBeanPostProcessor物件後,每次容器構造bean後都會執行ScheduledAnnotationBeanPostProcessor#postProcessAfterInitialization()

public Object postProcessAfterInitialization(final Object bean, String beanName) {
                //獲取具體的類
        final Class<?> targetClass = AopUtils.getTargetClass(bean);
                //獲取並遍歷該類的所有方法
        ReflectionUtils.doWithMethods(targetClass, new MethodCallback() {
            public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
                                 //獲取該方法的Scheduled註解
                Scheduled annotation = AnnotationUtils.getAnnotation(method, Scheduled.class);
                                 //如果方法沒有Scheduled註解,那就沒有必要執行了
                if (annotation != null) {
                    try {
                        ///省略一些程式碼
                        //獲取Scheduled註解的cron屬性值
                        String cron = annotation.cron();
                        if (!"".equals(cron)) {
                            Assert.isTrue(initialDelay == -1, "'initialDelay' not supported for cron triggers");
                            processedSchedule = true;
                            if (embeddedValueResolver != null) {
                                cron = embeddedValueResolver.resolveStringValue(cron);
                            }
                                                        //往registrar加一個定時任務
                            registrar.addCronTask(new CronTask(runnable, cron));
                        }
                        // 下面程式碼省略,具體讀者可以自行開啟IDE檢視這個類
                }
            }
        });

3. ScheduledTaskRegistrar類

從上面可以看出,ScheduledAnnotationBeanPostProcessor會遍歷spring 容器中所有bean的所有方法,如果發現有Scheduled註解的,就解析然後往registrar加入一個CronTask物件。所以我們再接著來分析registrar。從程式碼可以很直觀看到,它對應的類是ScheduledTaskRegistrar。我們來看一下它一些變數的定義

    private TaskScheduler taskScheduler;
    private ScheduledExecutorService localExecutor;
    private List<TriggerTask> triggerTasks;
    private List<CronTask> cronTasks;
    private List<IntervalTask> fixedRateTasks;
    private List<IntervalTask> fixedDelayTasks;
    private final Set<ScheduledFuture<?>> scheduledFutures = new LinkedHashSet<ScheduledFuture<?>>();

在前面的程式碼中,spring將各個Task加入到各種List中。但是各種Task只是一個抽象的類,java程式並不會識別他們並定時的執行任務。那是什麼時候講Task轉化成真正執行著的定時任務呢？
這裡我們先來看下ScheduledTaskRegistrar的初始化。由於它實現了InitializingBean介面，所以它構造完後會執行對應的afterPropertiesSet方法。

public void afterPropertiesSet() {
        scheduleTasks();
}
protected void scheduleTasks() {
        long now = System.currentTimeMillis();

        if (this.taskScheduler == null) {
            //構造一個單執行緒的執行緒池。所以所有的定時任務其實都是一個執行緒在跑。
            this.localExecutor = Executors.newSingleThreadScheduledExecutor();
            //具體定時任務的實現類，下面會講
            this.taskScheduler = new ConcurrentTaskScheduler(this.localExecutor);
        }
        if (this.triggerTasks != null) {
            for (TriggerTask task : triggerTasks) {
                this.scheduledFutures.add(this.taskScheduler.schedule(
                        task.getRunnable(), task.getTrigger()));
            }
        }
        if (this.cronTasks != null) {
            for (CronTask task : cronTasks) {
                        //遍歷task集合,然後將返回的scheduledFuture加入到scheduledFutures集合中
                this.scheduledFutures.add(this.taskScheduler.schedule(
                        task.getRunnable(), task.getTrigger()));
            }
        }
        //為了篇幅,繼續省略一些程式碼
}

這裡我們看到了這個類怎麼構造ScheduledFuture並加入到scheduledFutures集合中的。下面繼續來看構造ScheduledFuture的關鍵類ConcurrentTaskScheduler

4. ConcurrentTaskScheduler類

首先我們先隨便來看一個方法:

public ScheduledFuture scheduleAtFixedRate(Runnable task, Date startTime, long period) {
        long initialDelay = startTime.getTime() - System.currentTimeMillis();
        try {
              //this.scheduledExecutor就是前面定義的那個單執行緒池
              //errorHandlingTask(task, true) 是對任務進行封裝,如果任務丟擲異常,我們可以按自己選擇的方式來處理異常。第二個引數為true表示就算髮生異常也會繼續執行下一次任務。
            return this.scheduledExecutor.scheduleAtFixedRate(
                    errorHandlingTask(task, true), initialDelay, period, TimeUnit.MILLISECONDS);
        }
        catch (RejectedExecutionException ex) {
            throw new TaskRejectedException("Executor [" + this.scheduledExecutor + "] did not accept task: " + task, ex);
        }
    }

這裡可以看出,spring scheduler底層其實用的還是java的ScheduledExecutorService配置定時任務的。

再來看下最常用的cron方式配置的定時任務是怎麼下發的:

public ScheduledFuture schedule(Runnable task, Trigger trigger) {
        try {
            ErrorHandler errorHandler =
                    (this.errorHandler != null ? this.errorHandler : TaskUtils.getDefaultErrorHandler(true));
            //將構造ScheduledFuture的權利交給了ReschedulingRunnable類
            return new ReschedulingRunnable(task, trigger, this.scheduledExecutor, errorHandler).schedule();
        }
        catch (RejectedExecutionException ex) {
            throw new TaskRejectedException("Executor [" + this.scheduledExecutor + "] did not accept task: " + task, ex);
        }
    }

5. ReschedulingRunnable類

繼續跟蹤檢視ReschedulingRunnable類的幾個重要方法:

public ScheduledFuture schedule() {
        synchronized (this.triggerContextMonitor) {
            //根據cron表示式計算出下一次要執行的時間
            this.scheduledExecutionTime = this.trigger.nextExecutionTime(this.triggerContext);
            if (this.scheduledExecutionTime == null) {
                return null;
            }
            long initialDelay = this.scheduledExecutionTime.getTime() - System.currentTimeMillis();
              //用java自帶的ScheduledExecutorService下發一個延遲任務(過xxxms後執行)
            this.currentFuture = this.executor.schedule(this, initialDelay, TimeUnit.MILLISECONDS);
            return this;
        }
    }

    @Override
    public void run() {
//真正開始跑
        Date actualExecutionTime = new Date();
        //執行父類的run方法
        super.run();
        Date completionTime = new Date();
        synchronized (this.triggerContextMonitor) {
            this.triggerContext.update(this.scheduledExecutionTime, actualExecutionTime, completionTime);
          //如果沒有收到結束通知,繼續看下一次什麼時候執行
            if (!this.currentFuture.isCancelled()) {
                schedule();
            }
        }
    }

這是ReschedulingRunnable類的兩個關鍵方法,schedule返回一個ScheduledFuture物件。他其實就是分析cron表示式後獲取到下一次要執行的時間,然後交給ScheduledExecutorService去下發。
到執行的時間後,會執行run方法。run方法執行到後面,會繼續呼叫schedule()方法。然後再計算下次執行的時候,然後下發。這兩個方法一直迴圈,就構成了一個定時任務。

6.1將任務排程器快取多執行緒

前面我們可以看到,這裡的排程任務是多執行緒執行的,spring也提供了將單執行緒換成多執行緒執行的地方。我們是先看原始碼，還是ScheduledAnnotationBeanPostProcessor這個類。

public void onApplicationEvent(ContextRefreshedEvent event) {
        if (event.getApplicationContext() != this.applicationContext) {
            return;
        }
 //獲取所有我們自己定義的SchedulingConfigurer介面實現類
        Map<String, SchedulingConfigurer> configurers =
                this.applicationContext.getBeansOfType(SchedulingConfigurer.class);

        if (this.scheduler != null) {
            this.registrar.setScheduler(this.scheduler);
        }
 //這裡會將ScheduledTaskRegistrar傳入這個方法,這樣我們可以定義一個類,繼承SchedulingConfigurer介面,到這裡拿到ScheduledTaskRegistrar就可以修改執行緒池的實現了
        for (SchedulingConfigurer configurer : configurers.values()) {
            configurer.configureTasks(this.registrar);
        }
//後面省略
    }

spring的生命週期中,會自動呼叫onApplicationEvent這個方法。接著拿到SchedulingConfigurer的所有實現類,呼叫configureTasks方法,同時還會傳入ScheduledTaskRegistrar物件。這個ScheduledTaskRegistrar例項是spring實現任務排程的關鍵所在!!!所以只要我們修改一下它的執行緒池實現(預設是單執行緒池),就可以將任務排程變成多執行緒了。下面上程式碼:

@Component
public class ScheduleConfig implements SchedulingConfigurer {
    @Override
    public void configureTasks(ScheduledTaskRegistrar taskRegistrar) {
        ThreadPoolTaskScheduler threadPoolTaskScheduler = new ThreadPoolTaskScheduler();
        threadPoolTaskScheduler.setPoolSize(10);
        threadPoolTaskScheduler.initialize();

        taskRegistrar.setTaskScheduler(threadPoolTaskScheduler);
        taskRegistrar.getScheduler().schedule(new Runnable() {
            @Override
            public void run() {
                System.out.println("hello");
            }
        }, new CronTrigger("0 0/5 * * * ?"));
    }
}

這裡將排程的執行緒池調到了10個執行緒。

6.2 手動注入ThreadPoolTaskScheduler來實現多執行緒

<task:scheduler id="scheduler" pool-size="10"/>

在spring配置檔案中加上上面的程式碼,spring啟動的時候會往容器注入一個ThreadPoolTaskScheduler 例項。這個例項執行緒池大小配置為10。這樣,spring啟動的時候就會自動選擇這個例項來作為ScheduledTaskRegistrar的TaskScheduler的實現。如果沒在容器裡面發現任何TaskScheduler例項,就會new一個單執行緒的執行緒池。

if (this.registrar.hasTasks() && this.registrar.getScheduler() == null) {
            Map<String, ? super Object> schedulers = new HashMap<String, Object>();
            schedulers.putAll(applicationContext.getBeansOfType(TaskScheduler.class));
            schedulers.putAll(applicationContext.getBeansOfType(ScheduledExecutorService.class));
//沒在容器中發現任何定時任務的執行緒池,就用預設的單執行緒池
            if (schedulers.size() == 0) {
                // do nothing -> fall back to default scheduler
            }
//發現了一個,就直接用它了
            else if (schedulers.size() == 1) {
                this.registrar.setScheduler(schedulers.values().iterator().next());
            }
//如果超過兩個會報錯哦!
            else if (schedulers.size() >= 2){
                throw new IllegalStateException(
                        "More than one TaskScheduler and/or ScheduledExecutorService  " +
                        "exist within the context. Remove all but one of the beans; or " +
                        "implement the SchedulingConfigurer interface and call " +
                        "ScheduledTaskRegistrar#setScheduler explicitly within the " +
                        "configureTasks() callback. Found the following beans: " + schedulers.keySet());
            }
        }

7. 結論

分析到這裡我們其實可以得出一些結論:

• spring scheduler底層使用的還是juc中的ScheduledExecutorService類。
• 預設情況下,所有的定時任務都是放在一個執行緒跑的。也就是說,如果這個執行緒執行某個任務崩潰了或者執行某個任務的時候卡住了，那其他的定時任務也就都無法執行了。
• 通過配置可以將單執行緒執行所有排程變成多執行緒執行所有排程任務。
• TaskScheduler 介面是spring排程的核心介面,排程底層都用到了它(確切的說是它的一些實現類)。
• 由於spring使用java**ScheduledExecutorService**配置定時任務的時候對任務做了封裝,所以即使執行的任務發生了我們沒有捕獲的異常，也不會導致執行緒崩潰。因此,導致spring scheduler定時任務全部停止的原因很可能是因為執行某個定時任務的時候卡住了。

總結到這裡,再聯想到最近往vos加了個定時任務,裡面有用到httpclient傳送http請求。猜想可能是這裡卡住了。所以接著去了解了下apache的httpclient這個包。

二、httpclient 輸入流未設定超時時間引起的阻塞

網上查了一下,httpclient的超時其實有兩種,一種是連線超時(connect timeout)，一種是讀超時(socket timeout)。
httpclient的預設讀時間是設定為0,也就是永遠不超時的意思,這樣就導致讀的時候因為出現某個問題執行緒阻塞在讀這裡。通過jstack命令檢視執行緒堆疊我們也可以確認執行緒確實阻塞在socket讀那裡了。

"org.springframework.jms.listener.DefaultMessageListenerContainer#7-1" prio=10 tid=0x00007f345127d800 nid=0x5b4f0 runnable [0x00007f34753d1000]  
   java.lang.Thread.State: RUNNABLE  
    at java.net.SocketInputStream.socketRead0(Native Method) 
    at java.net.SocketInputStream.read(SocketInputStream.java:150)  
    at java.net.SocketInputStream.read(SocketInputStream.java:121)  
    at org.apache.http.impl.io.AbstractSessionInputBuffer.fillBuffer(AbstractSessionInputBuffer.java:130)  
    at org.apache.http.impl.io.SocketInputBuffer.fillBuffer(SocketInputBuffer.java:127)  
    at org.apache.http.impl.io.AbstractSessionInputBuffer.readLine(AbstractSessionInputBuffer.java:233)  
    at org.apache.http.impl.io.ChunkedInputStream.getChunkSize(ChunkedInputStream.java:220)  
    at org.apache.http.impl.io.ChunkedInputStream.nextChunk(ChunkedInputStream.java:183)  
    at org.apache.http.impl.io.ChunkedInputStream.read(ChunkedInputStream.java:152)  
    at org.apache.http.conn.EofSensorInputStream.read(EofSensorInputStream.java:138)  
    at sun.nio.cs.StreamDecoder.readBytes(StreamDecoder.java:283)  
    at sun.nio.cs.StreamDecoder.implRead(StreamDecoder.java:325)  
    at sun.nio.cs.StreamDecoder.read(StreamDecoder.java:177)  
    - locked <0x000000070346ce70> (a java.io.InputStreamReader)  
    at java.io.InputStreamReader.read(InputStreamReader.java:184)  
    at java.io.Reader.read(Reader.java:140)  
    at org.apache.http.util.EntityUtils.toString(EntityUtils.java:161)  

我們可以通過設定讀超時時間來解決這個問題:

HttpClient client = new HttpClient();
client.getHttpConnectionManager().getParams().setConnectionTimeout(5000);
client.getHttpConnectionManager().getParams().setSoTimeout(5000);

這裡設定了5秒的連線超時時間和讀超時時間,也就是說,阻塞5秒後如果還沒有得到響應,就會放棄繼續讀,讓執行緒可以往下執行。