@Override
    public void start() {
        masterUrl = mqConfig.getTubeMasterList();
        Preconditions.checkArgument(StringUtils.isNotEmpty(masterUrl), "no tube masterUrlList specified");
        topic = mqConfig.getTubeTopic();
        Preconditions.checkArgument(StringUtils.isNotEmpty(topic), "no tube topic specified");
        fetchThreadCnt = mqConfig.getTubeThreadNum();
        Preconditions.checkArgument(StringUtils.isNotEmpty(mqConfig.getTubeConsumerGroupName()),
                "no tube consumer groupName specified");

        initConsumer();

        Thread[] fetchRunners = new Thread[fetchThreadCnt];
        for (int i = 0; i < fetchThreadCnt; i++) {
            fetchRunners[i] = new Thread(new Fetcher(pullConsumer, topic), "TubeConsume_Fetcher_Thread_" + i);
            fetchRunners[i].start();
        }
    }
Constructor
@param insertServiceList
@param storeConfig
@param mqConfig
inlong-audit/audit-store/src/main/java/org/apache/inlong/audit/store/service/consume/TubeConsume.java