聊聊chronos的DeleteBgWorker

序

本文主要研究一下chronos的DeleteBgWorker

DeleteBgWorker

DDMQ/carrera-chronos/src/main/java/com/xiaojukeji/chronos/workers/DeleteBgWorker.java

public class DeleteBgWorker {
    private static final Logger LOGGER = LoggerFactory.getLogger(DeleteBgWorker.class);

    private static final DeleteConfig DELETE_CONFIG = ConfigManager.getConfig().getDeleteConfig();
    private static final int SAVE_HOURS_OF_DATA = DELETE_CONFIG.getSaveHours();
    private static final long INITIAL_DELAY_MINUTES = 1; // 1 分钟
    private static final long PERIOD_MINUTES = 10;       // 10 分钟

    private static volatile DeleteBgWorker instance = null;
    private static final SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    /**
     * 2017/10/13 00:00:00
     */
    private static final long MIN_TIMESTAMP = 1507824000;

    private static final ScheduledExecutorService SCHEDULE = new ScheduledThreadPoolExecutor(1,
            new BasicThreadFactory.Builder().namingPattern("delete-bg-worker-schedule-%d").daemon(true).build());

    private DeleteBgWorker() {
    }

    public void start() {
        SCHEDULE.scheduleAtFixedRate(() -> {
            byte[] beginKey = String.valueOf(MIN_TIMESTAMP).getBytes(Charsets.UTF_8);

            final long seekTimestampInSecond = MetaService.getSeekTimestamp();
            byte[] endKey = String.valueOf(seekTimestampInSecond - SAVE_HOURS_OF_DATA * 60 * 60).getBytes(Charsets.UTF_8);
            deleteRange(beginKey, endKey);
        }, INITIAL_DELAY_MINUTES, PERIOD_MINUTES, TimeUnit.MINUTES);
        LOGGER.info("DeleteBgWorker has started, initialDelayInMinutes:{}", INITIAL_DELAY_MINUTES);
    }

    private void deleteRange(final byte[] beginKey, final byte[] endKey) {
        LOGGER.info("deleteRange start, beginKey:{}, endKey:{}", new String(beginKey), new String(endKey));
        final long start = System.currentTimeMillis();
        RDB.deleteFilesInRange(CFManager.CFH_DEFAULT, beginKey, endKey);
        LOGGER.info("deleteRange end, beginKey:{}({}), endKey:{}({}), cost:{}ms",
                new String(beginKey), formatter.format(Long.parseLong(new String(beginKey)) * 1000),
                new String(endKey), formatter.format(Long.parseLong(new String(endKey)) * 1000),
                System.currentTimeMillis() - start);
    }

    public void stop() {
        SCHEDULE.shutdownNow();
        while (!SCHEDULE.isShutdown()) {
            LOGGER.info("DeleteBgWorker is shutting down!");
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                LOGGER.info("DeleteBgWorker was forced to shutdown, err:{}", e.getMessage(), e);
            }
        }
        LOGGER.info("DeleteBgWorker was shutdown!");
    }

    public static DeleteBgWorker getInstance() {
        if (instance == null) {
            synchronized (DeleteBgWorker.class) {
                if (instance == null) {
                    instance = new DeleteBgWorker();
                }
            }
        }
        return instance;
    }
}

• DeleteBgWorker提供了静态方法getInstance来获取或创建单例，该类提供了start、stop两个方法；start方法会往SCHEDULE注册一个调度任务，每隔PERIOD_MINUTES执行一次，它主要执行deleteRange方法；deleteRange主要是执行RDB.deleteFilesInRange(CFManager.CFH_DEFAULT, beginKey, endKey)，它会从MetaService获取seekTimestamp来计算endKey；stop方法则是关闭SCHEDULE

deleteFilesInRange

DDMQ/carrera-chronos/src/main/java/com/xiaojukeji/chronos/db/RDB.java

public class RDB {
    private static final org.slf4j.Logger LOGGER = LoggerFactory.getLogger(RDB.class);

    static RocksDB DB;

    //......

    public static boolean deleteFilesInRange(final ColumnFamilyHandle cfh, final byte[] beginKey,
                                             final byte[] endKey) {
        try {
            DB.deleteRange(cfh, beginKey, endKey);
            LOGGER.debug("succ delete range, columnFamilyHandle:{}, beginKey:{}, endKey:{}",
                    cfh.toString(), new String(beginKey), new String(endKey));
        } catch (RocksDBException e) {
            LOGGER.error("error while delete range, columnFamilyHandle:{}, beginKey:{}, endKey:{}, err:{}",
                    cfh.toString(), new String(beginKey), new String(endKey), e.getMessage(), e);
            return false;
        }
        return true;
    }

    //......
}

• deleteFilesInRange方法主要是执行DB.deleteRange(cfh, beginKey, endKey)

MetaService

DDMQ/carrera-chronos/src/main/java/com/xiaojukeji/chronos/services/MetaService.java

public class MetaService {
    private static final Logger LOGGER = LoggerFactory.getLogger(MetaService.class);

    private static volatile long seekTimestamp = -1;
    private static volatile long zkSeekTimestamp = -1;
    private static volatile Map<String, Long> zkQidOffsets = new ConcurrentHashMap<>();
    private static final DbConfig dbConfig = ConfigManager.getConfig().getDbConfig();

    private static final ScheduledExecutorService SCHEDULER = new ScheduledThreadPoolExecutor(1,
            new BasicThreadFactory.Builder().namingPattern("offset-seektimestamp-schedule-%d").daemon(true).build());

    public static void load() {
        final long start = System.currentTimeMillis();
        if (seekTimestamp == -1) {
            seekTimestamp = loadSeekTimestampFromFile();
        }
        final long cost = System.currentTimeMillis() - start;
        LOGGER.info("succ load seekTimestamp, seekTimestamp:{}, cost:{}ms", seekTimestamp, cost);

        SCHEDULER.scheduleWithFixedDelay(() -> {
            // 如果是master则拉取并上报zk offset和seekOffset
            if (MasterElection.isMaster()) {
                MqConsumeStatService.getInstance().uploadOffsetsToZk();
                uploadSeekTimestampToZk();
            }
        }, 5, 1, TimeUnit.SECONDS);
    }

    private static long loadSeekTimestampFromFile() {
        String seekTimestampStr = FileIOUtils.readFile2String(dbConfig.getSeekTimestampPath());
        if (StringUtils.isBlank(seekTimestampStr)) {
            final long initSeekTimestamp = TsUtils.genTS();
            boolean result = FileIOUtils.writeFileFromString(dbConfig.getSeekTimestampPath(), String.valueOf(initSeekTimestamp));
            if (result) {
                LOGGER.info("init seekTimestamp and succ save, current seekTimestamp:{}", initSeekTimestamp);
            } else {
                LOGGER.error("init seekTimestamp and fail to save, current seekTimestamp:{}", initSeekTimestamp);
            }
            return initSeekTimestamp;
        }
        LOGGER.info("succ load seekTimestamp from file, seekTimestamp:{}", Long.parseLong(seekTimestampStr));
        return Long.parseLong(seekTimestampStr);
    }

    public static long getSeekTimestamp() {
        return seekTimestamp;
    }

    /**
     * 此处的lock不能去掉
     * 判断消息超时时
     */
    public static void nextSeekTimestamp() {
        Batcher.lock.lock();
        try {
            seekTimestamp++;
            boolean result = FileIOUtils.writeFileFromString(dbConfig.getSeekTimestampPath(), String.valueOf(seekTimestamp));
            if (result) {
                LOGGER.info("incr seekTimestamp and succ save, next seekTimestamp:{}", seekTimestamp);
            } else {
                LOGGER.error("incr seekTimestamp and fail to save, next seekTimestamp:{}", seekTimestamp);
            }
        } finally {
            Batcher.lock.unlock();
        }
    }

    public static void uploadSeekTimestampToZk() {
        String seekTimestampStr = String.valueOf(MetaService.getSeekTimestamp());
        ZkUtils.createOrUpdateValue(Constants.SEEK_TIMESTAMP_ZK_PATH, seekTimestampStr);
        LOGGER.debug("upload seekTimestamp to zk, seekTimestamp:{}", seekTimestampStr);
    }

    public static Map<String, Long> getZkQidOffsets() {
        return zkQidOffsets;
    }

    public static void setZkQidOffsets(Map<String, Long> zkQidOffsets) {
        MetaService.zkQidOffsets = zkQidOffsets;
    }

    public static long getZkSeekTimestamp() {
        return zkSeekTimestamp;
    }

    public static void setZkSeekTimestamp(long zkSeekTimestamp) {
        MetaService.zkSeekTimestamp = zkSeekTimestamp;
    }
}

• MetaService提供了load、getSeekTimestamp、nextSeekTimestamp、uploadSeekTimestampToZk等方法；load方法在seekTimestamp为-1时执行loadSeekTimestampFromFile，之后注册一个定时任务每隔1秒，判断是否是master，如果是则执行MqConsumeStatService.getInstance().uploadOffsetsToZk()以及uploadSeekTimestampToZk；nextSeekTimestamp方法会先更新内存的seekTimestamp，然后使用FileIOUtils.writeFileFromString将其值写入文件

小结

DeleteBgWorker提供了静态方法getInstance来获取或创建单例，该类提供了start、stop两个方法；start方法会往SCHEDULE注册一个调度任务，每隔PERIOD_MINUTES执行一次，它主要执行deleteRange方法；deleteRange主要是执行RDB.deleteFilesInRange(CFManager.CFH_DEFAULT, beginKey, endKey)，它会从MetaService获取seekTimestamp来计算endKey；stop方法则是关闭SCHEDULE

doc

• carrera-chronos