背景

團(tuán)隊(duì)小伙伴突然找到我們團(tuán)隊(duì)說(shuō)，不得了了，線上的RocketMQ又出現(xiàn)了消費(fèi)停滯，怎么辦？ 要不要我們先重啟一下

其實(shí)早在之前也出現(xiàn)過(guò)一次，當(dāng)時(shí)為了快速恢復(fù)業(yè)務(wù)的，就直接重啟解決的。

這次因?yàn)槎逊e量不多，所以想對(duì)運(yùn)行環(huán)境進(jìn)行一些環(huán)境快照保留。所以就和業(yè)務(wù)方溝通是否看見(jiàn)接受短暫的消息延時(shí)

得到肯定答案后就是放手干吧！

問(wèn)題定位

首先我們要確定業(yè)務(wù)反饋的是否屬實(shí)，所以需要去RocketMQ dashboard上看看具體的消費(fèi)進(jìn)度。

圖片

可以看到consumer中并不是全部的queue消費(fèi)都堆積了，只有一個(gè)queuq消息堆積了。

這個(gè)消費(fèi)者訂閱的topic是分區(qū)有序的，正常來(lái)說(shuō)分區(qū)有序，如果某個(gè)分區(qū)的消息單條消息出現(xiàn)了消費(fèi)異常，必須要等這條消息消費(fèi)成功(或者是重試結(jié)束)后才能繼續(xù)消費(fèi)后面的消息。

有時(shí)候會(huì)因?yàn)檫@個(gè)原因出現(xiàn)消息堆積是正常的，但是業(yè)務(wù)對(duì)消息重試進(jìn)行了合理的設(shè)置，設(shè)置的重試次數(shù)比較合理，不會(huì)出現(xiàn)長(zhǎng)時(shí)間的堆積。

RocketMQ的消費(fèi)線程

一般出現(xiàn)這種問(wèn)題很明顯就是線程出現(xiàn)了死鎖或者僵死之類(lèi)的情況。

熟悉RocketMQ的都知道RocketMQ消費(fèi)消息主要是依賴(lài)1個(gè)線程1個(gè)線程池。

1. 以PullMessageService開(kāi)頭的線程, 主要用來(lái)拉去消息

圖片

2. 以ConsumeMessageThread開(kāi)頭的線程(實(shí)際是一個(gè)線程池),主要用來(lái)執(zhí)行消費(fèi)邏輯。

圖片

直到了RocketMQ的消費(fèi)線程模型后我們就好解決了。我們直接通過(guò)jstack命令查看線程的堆棧信息。

線程快照分析

我們直接通過(guò)jstack命令生成線程快照。

jstack <pid> > thread_dump_$(date +%Y%m%d_%H%M%S).txt

pid 和后面的 thread_dump_$(date +%Y%m%d_%H%M%S).txt自己隨便取個(gè)名字就行。自己記得就行。

由于應(yīng)用運(yùn)行在pod中，生成了我們就下載到本地。

我們自己看還是比較難分析出分體。這里我們直接使用一個(gè)在線的網(wǎng)站進(jìn)行線程快照的分析。

fastthread

fastthread是一個(gè)在線的線程快照分析工具，可以直接將線程快照上傳到這個(gè)網(wǎng)站進(jìn)行分析。

圖片

我們上傳我們下載的線程快照文件。

然后進(jìn)行線程分析：

圖片

很快定位到阻塞其他線程的代碼。

這里的代碼被我打碼了。

arhtas

如果我們使用arthas也可以很方便的找到阻塞的線程。

thread -b

arthas 提供了thread -b， 一鍵找出那個(gè)罪魁禍?zhǔn)住?/p>

問(wèn)題元兇找到

通過(guò)阻塞代碼我們很快定位到是由于線程池使用不當(dāng)導(dǎo)致的阻塞。

線程池使用不當(dāng)

什么情況下會(huì)出現(xiàn)線程池使用不當(dāng)導(dǎo)致的"死鎖"呢？

我們看看下面的demo：

public class XiaoZouExample {

    public static void main(String[] args) {

        ExecutorService executor = new ThreadPoolExecutor(2, 5, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());


        // 提交兩個(gè)外部任務(wù)
        for (int i = 0; i < 2; i++) {
            executor.submit(new OuterTask(executor));
        }

        // 等待一段時(shí)間后關(guān)閉線程池
        try {
            Thread.sleep(10000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        executor.shutdown();

    }

    static class OuterTask implements Runnable {

        private final ExecutorService executor;

        public OuterTask(ExecutorService executor) {
            this.executor = executor;
        }

        @Override
        public void run() {
            System.out.println("小奏技術(shù) Outer task started by thread: " + Thread.currentThread().getName());

            // 創(chuàng)建一個(gè)Future來(lái)等待內(nèi)部任務(wù)的結(jié)果
            Future<?> future = executor.submit(new InnerTask());

            try {
                // 等待內(nèi)部任務(wù)完成
                future.get();
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }

            System.out.println("小奏技術(shù) Outer task finished by thread: " + Thread.currentThread().getName());
        }
    }

    static class InnerTask implements Runnable {
        @Override
        public void run() {
            System.out.println("小奏技術(shù) Inner task started by thread: " + Thread.currentThread().getName());
            try {
                // 模擬長(zhǎng)時(shí)間運(yùn)行的任務(wù)
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("小奏技術(shù) Inner task finished by thread: " + Thread.currentThread().getName());
        }
    }

}

• 運(yùn)行結(jié)果

圖片

可以看到?jīng)]有任何任務(wù)執(zhí)行完成，線程池一直處于被阻塞狀態(tài)。

核心原因就是首先線程池的核心線程數(shù)是2，核心線程用來(lái)執(zhí)行2個(gè)任務(wù)，用完了所有線程。

然后在核心線程執(zhí)行的2個(gè)任務(wù)中又用原來(lái)的線程池進(jìn)行執(zhí)行任務(wù)，這時(shí)候因?yàn)闆](méi)有線程可以去執(zhí)行任務(wù)了，所以會(huì)添加到阻塞隊(duì)列中等待核心線程執(zhí)行完任務(wù)后再執(zhí)行。

但是核心線程想要釋放任務(wù)又必須等待這兩個(gè)子任務(wù)執(zhí)行完，這樣就形成了一個(gè)死鎖。

解決方案

解決方式有多種，最簡(jiǎn)單的方式可以考慮不要使用隊(duì)列，直接使用SynchronousQueue。

ExecutorService executor = new ThreadPoolExecutor(2, 5, 0L, TimeUnit.MILLISECONDS, new SynchronousQueue<>());

讓多的任務(wù)直接通過(guò)主線程執(zhí)行或者丟棄任務(wù)。

當(dāng)然最好的方式就是避免這種情況的發(fā)生，合理的使用線程池，不要線程池中的任務(wù)還要使用這個(gè)線程池去執(zhí)行任務(wù)。

這種情況是比較難避免的，因?yàn)楝F(xiàn)在大部分業(yè)務(wù)開(kāi)發(fā)都是隱式使用線程池，自己也不知道自己用的哪個(gè)線程池。

比如spring的@Async注解，@Scheduled注解等等。

總結(jié)

線程池的使用還是要盡量消息，要避免出現(xiàn)線程池中的任務(wù)繼續(xù)使用該線程池去執(zhí)行任務(wù)，出現(xiàn)死鎖。

也可以考慮對(duì)線程池進(jìn)行監(jiān)控，避免出現(xiàn)大量任務(wù)阻塞。

這個(gè)問(wèn)題想要復(fù)現(xiàn)需要大量任務(wù)并且超過(guò)核心線程數(shù)才能復(fù)現(xiàn)，還是比較難復(fù)現(xiàn)的，只有線上大流量的時(shí)候才能復(fù)現(xiàn)。