如何做一个实时的业务统计的监控？比如分钟级？也就是每分钟可以快速看到业务的变化趋势，及可以做一些简单的分组查询？
哎，你可能说很简单了，直接从数据库 count 就可以了！ 你是对的。
但如果不允许你使用db进行count呢？因为线上数据库资源可是很宝贵的哦，你这一count可能会给db带来灾难了。
那不然咋整？
没有db，我们还有其他数据源嘛，比如: 消息队列？埋点数据？ 本文将是基于该前提而行。
做监控，尽量不要侵入业务太多！所以有一个消息中间件是至关重要的。针对大数据系统,一般是: kafka 或者 类kafka. (如本文基础 loghub)
有了消息中间件，如何进行分钟级监控？ 这个应该就很简单了吧。不过如果要自己实现，其实坑也不少的！
如果自己实现计数，那么你可能需要做以下几件事:
1. 每消费一个消息，你需要一个累加器;
2. 每隔一个周期，你可能需要一个归档操作;
3. 你可能需要考虑各种并发安全问题;
4. 你可能需要考虑种性能问题;
5. 你可能需要考虑各种机器故障问题;
6. 你可能需要考虑各种边界值问题;
哎，其实没那么难。时间序列数据库，就专门为这类事情而生！如OpenTSDB: http://opentsdb.net/overview.html
可以说，TSDB 是这类应用场景的杀手锏。或者基于流计算框架: 如flink, 也是很轻松完成的事。但是不是本文的方向，略过！
本文是基于 loghub 的现有数据，进行分钟级统计后，入库 mysql 中，从而支持随时查询。（因loghub每次查询都是要钱的，所以，不可能直接查询）
loghub 数据结构如: 2019-07-10 10:01:11,billNo,userId,productCode,...
由于loghub提供了很多强大的查询统计功能，所以我们可以直接使用了。
核心功能就是一个统计sql,还是比较简单的。但是需要考虑的点也不少，接下来，将为看官们奉上一个完整的解决方案！
撸代码去！
1. 核心统计任务实现类 MinuteBizDataCounterTask
import com.aliyun.openservices.log.Client;
import com.aliyun.openservices.log.common.LogContent;
import com.aliyun.openservices.log.common.LogItem;
import com.aliyun.openservices.log.common.QueriedLog;
import com.aliyun.openservices.log.exception.LogException;
import com.aliyun.openservices.log.response.GetLogsResponse;
import com.my.service.statistics.StatisticsService;
import com.my.entity.BizDataStatisticsMin;
import com.my.model.LoghubQueryCounterOffsetModel;
import com.my.util.loghub.LogHubProperties;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.ArrayList;
import java.util.List;
/**
* 基于loghub 的分钟级 统计任务
*/
@Component
@Slf4j
public class MinuteBizDataCounterTask implements Runnable {
@Resource
private LogHubProperties logHubProperties;
@Resource
private StatisticsService statisticsService;
@Resource(name = "defaultOffsetQueryTaskCallback")
private DefaultOffsetQueryTaskCallbackImpl defaultOffsetQueryTaskCallback;
/**
* loghub 客户端
*/
private volatile Client mClient;
/**
* 过滤的topic
*/
private static final String LOGHUB_TOPIC = "topic_test";
/**
* 单次扫描loghub最大时间 间隔分钟数
*/
@Value("${loghub.offset.counter.perScanMaxMinutesGap}")
private Integer perScanMaxMinutesGap;
/**
* 单次循环最大数
*/
@Value("${loghub.offset.counter.perScanMaxRecordsLimit}")
private Integer perScanMaxRecordsLimit;
/**
* 构造必要实例信息
*/
public ProposalPolicyBizDataCounterTask() {
}
@Override
public void run() {
if(mClient == null) {
this.mClient = new Client(logHubProperties.getEndpoint(),
logHubProperties.getAccessKeyId(), logHubProperties.getAccessKey());
}
while (!Thread.interrupted()) {
try {
updateLastMinutePolicyNoCounter();
Thread.sleep(60000);
}
catch (InterruptedException e) {
log.error("【分钟级统计task】, sleep 中断", e);
Thread.currentThread().interrupt();
}
catch (Exception e) {
// 注意此处可能有风险，发生异常后将快速死循环
log.error("【分钟级统计task】更新异常", e);
try {
Thread.sleep(10000);
}
catch (InterruptedException ex) {
log.error("【分钟级统计task】异常，且sleep异常", ex);
Thread.currentThread().interrupt();
}
}
}
}
/**
* 更新最近的数据 （分钟级）
*
* @throws LogException loghub查询异常时抛出
*/
private void updateLastMinutePolicyNoCounter() throws LogException {
updateMinutePolicyNoCounter(null);
}
/**
* 更新最近的数据
*/
public Integer updateMinutePolicyNoCounter(LoghubQueryCounterOffsetModel specifyOffset) throws LogException {
// 1. 获取偏移量
// 2. 根据偏移量，判定是否可以一次性取完，或者多次获取更新
// 3. 从loghub中设置偏移量，获取统计数据，更新
// 4. 更新db数据统计值
// 5. 更新偏移量
// 6. 等待下一次更新
// 指定offset时，可能为补数据
final LoghubQueryCounterOffsetModel destOffset = enhanceQueryOffset(specifyOffset);
initSharedQueryOffset(destOffset, destOffset == specifyOffset);
Integer totalAffectNum = 0;
while (!isScanFinishOnDestination(destOffset)) {
// 完整扫描一次时间周期
calcNextSharedQueryOffset(destOffset);
while (true) {
calcNextInnerQueryOffset();
ArrayList<QueriedLog> logs = queryPerMinuteStatisticFromLoghubOnCurrentOffset();
Integer affectNum = handleMiniOffsetBatchCounter(logs);
totalAffectNum += affectNum;
log.info("【分钟级统计task】本次更新数据:{}, offset:{}", affectNum, getCurrentSharedQueryOffset());
if(!hasMoreDataOffset(logs.size())) {
rolloverOffsetAndCommit();
break;
}
}
}
log.info("【分钟级统计task】本次更新数据，总共:{}, destOffset:{}, curOffset:{}",
totalAffectNum, destOffset, getCurrentSharedQueryOffset());
rolloverOffsetAndCommit();
return totalAffectNum;
}
/**
* 处理一小批的统计数据
*
* @param logs 小批统计loghub数据
* @return 影响行数
*/
private Integer handleMiniOffsetBatchCounter(ArrayList<QueriedLog> logs) {
if (logs == null || logs.isEmpty()) {
return 0;
}
List<BizDataStatisticsMin> statisticsMinList = new ArrayList<>();
for (QueriedLog log1 : logs) {
LogItem getLogItem = log1.GetLogItem();
BizDataStatisticsMin statisticsMin1 = adaptStatisticsMinDbData(getLogItem);
statisticsMin1.setEventCode(PROPOSAL_FOUR_IN_ONE_TOPIC);
statisticsMin1.setEtlVersion(getCurrentScanTimeDuring() + ":" + statisticsMin1.getStatisticsCount());
statisticsMinList.add(statisticsMin1);
}
return statisticsService.batchUpsertPremiumStatistics(statisticsMinList, getCurrentOffsetCallback());
}
/**
* 获取共享偏移信息
*
* @return 偏移
*/
private LoghubQueryCounterOffsetModel getCurrentSharedQueryOffset() {
return defaultOffsetQueryTaskCallback.getCurrentOffset();
}
/**
* 判断本次是否扫描完成
*
* @param destOffset 目标偏移
* @return true:扫描完成, false: 未完成
*/
private boolean isScanFinishOnDestination(LoghubQueryCounterOffsetModel destOffset) {
return defaultOffsetQueryTaskCallback.getEndTime() >= destOffset.getEndTime();
}
/**
* 获取偏移提交回调器
*
* @return 回调实例
*/
private OffsetQueryTaskCallback getCurrentOffsetCallback() {
return defaultOffsetQueryTaskCallback;
}
/**
* 初始化共享的查询偏移变量
*
* @param destOffset 目标偏移
* @param isSpecifyOffset 是否是手动指定的偏移
*/
private void initSharedQueryOffset(LoghubQueryCounterOffsetModel destOffset, boolean isSpecifyOffset) {
// 整分花时间数据
Integer queryStartTime = destOffset.getStartTime();
if(queryStartTime % 60 != 0) {
queryStartTime = queryStartTime / 60 * 60;
}
// 将目标扫描时间终点 设置为起点，以备后续迭代
defaultOffsetQueryTaskCallback.initCurrentOffset(queryStartTime, queryStartTime,
destOffset.getOffsetStart(), destOffset.getLimit(),
destOffset.getIsNewStep(), isSpecifyOffset);
if(defaultOffsetQueryTaskCallback.getIsNewStep()) {
resetOffsetDefaultSettings();
}
}
/**
* 计算下一次统计偏移时间
*
* @param destOffset 目标偏移值
*/
private void calcNextSharedQueryOffset(LoghubQueryCounterOffsetModel destOffset) {
int perScanMaxSecondsGap = perScanMaxMinutesGap * 60;
if(destOffset.getEndTime() - defaultOffsetQueryTaskCallback.getStartTime() > perScanMaxSecondsGap) {
defaultOffsetQueryTaskCallback.setStartTime(defaultOffsetQueryTaskCallback.getEndTime());
int nextExpectEndTime = defaultOffsetQueryTaskCallback.getStartTime() + perScanMaxSecondsGap;
if(nextExpectEndTime > destOffset.getEndTime()) {
nextExpectEndTime = destOffset.getEndTime();
}
defaultOffsetQueryTaskCallback.setEndTime(nextExpectEndTime);
}
else {
defaultOffsetQueryTaskCallback.setStartTime(defaultOffsetQueryTaskCallback.getEndTime());
defaultOffsetQueryTaskCallback.setEndTime(destOffset.getEndTime());
}
resetOffsetDefaultSettings();
}
/**
* 重置偏移默认配置
*/
private void resetOffsetDefaultSettings() {
defaultOffsetQueryTaskCallback.setIsNewStep(true);
defaultOffsetQueryTaskCallback.setOffsetStart(0);
defaultOffsetQueryTaskCallback.setLimit(0);
}
/**
* 计算下一次小偏移，此种情况应对 一次外部偏移未查询完成的情况
*/
private void calcNextInnerQueryOffset() {
defaultOffsetQueryTaskCallback.setIsNewStep(false);
// 第一次计算时，limit 为0, 所以得出的 offsetStart 也是0
defaultOffsetQueryTaskCallback.setOffsetStart(
defaultOffsetQueryTaskCallback.getOffsetStart() + defaultOffsetQueryTaskCallback.getLimit());
defaultOffsetQueryTaskCallback.setLimit(perScanMaxRecordsLimit);
}
/**
* 获取当前循环的扫描区间
*
* @return 15567563433-1635345099 区间
*/
private String getCurrentScanTimeDuring() {
return defaultOffsetQueryTaskCallback.getStartTime() + "-" + defaultOffsetQueryTaskCallback.getEndTime();
}
/**
* 从loghub查询每分钟的统计信息
*
* @return 查询到的统计信息
* @throws LogException loghub 异常时抛出
*/
private ArrayList<QueriedLog> queryPerMinuteStatisticFromLoghubOnCurrentOffset() throws LogException {
// 先按保单号去重，再进行计数统计
String countSql = "* | split(bizData, ',')[5] policyNo, bizData GROUP by split(bizData, ',')[5] " +
" | select count(1) as totalCountMin, " +
"split(bizData, ',')[2] as productCode," +
"split(bizData, ',')[3] as schemaCode," +
"split(bizData, ',')[4] as channelCode," +
"substr(split(bizData, ',')[1], 1, 16) as myDateTimeMinute " +
"group by substr(split(bizData, ',')[1], 1, 16), split(bizData, ',')[2],split(bizData, ',')[3], split(bizData, ',')[4],split(bizData, ',')[7], split(bizData, ',')[8]";
countSql += " limit " + defaultOffsetQueryTaskCallback.getOffsetStart() + "," + defaultOffsetQueryTaskCallback.getLimit();
GetLogsResponse countResponse = mClient.GetLogs(logHubProperties.getProjectName(), logHubProperties.getBizCoreDataLogStore(),
defaultOffsetQueryTaskCallback.getStartTime(), defaultOffsetQueryTaskCallback.getEndTime(),
LOGHUB_TOPIC, countSql);
if(!countResponse.IsCompleted()) {
log.error("【分钟级统计task】扫描获取到未完整的数据，请速检查原因，offSet:{}", getCurrentSharedQueryOffset());
}
return countResponse.GetLogs() == null
? new ArrayList<>()
: countResponse.GetLogs();
}
/**
* 根据上一次返回的记录数量，判断是否还有更多数据
*
* @param lastGotRecordsCount 上次返回的记录数 (数据量大于最大数说明还有未取完数据)
* @return true: 是还有更多数据应该再循环获取, false: 无更多数据结束本期任务
*/
private boolean hasMoreDataOffset(int lastGotRecordsCount) {
return lastGotRecordsCount >= defaultOffsetQueryTaskCallback.getLimit();
}
/**
* 加强版的 offset 优先级: 指定偏移 -> 基于缓存的偏移 -> 新生成偏移标识
*
* @param specifyOffset 指定偏移(如有)
* @return 偏移标识
*/
private LoghubQueryCounterOffsetModel enhanceQueryOffset(LoghubQueryCounterOffsetModel specifyOffset) {
if(specifyOffset != null) {
return specifyOffset;
}
LoghubQueryCounterOffsetModel offsetBaseOnCache = getNextOffsetBaseOnCache();
if(offsetBaseOnCache != null) {
return offsetBaseOnCache;
}
return generateNewOffset();
}
/**
* 基于缓存获取一下偏移标识
*
* @return 偏移
*/
private LoghubQueryCounterOffsetModel getNextOffsetBaseOnCache() {
LoghubQueryCounterOffsetModel offsetFromCache = defaultOffsetQueryTaskCallback.getCurrentOffsetFromCache();
if(offsetFromCache == null) {
return null;
}
LocalDateTime now = LocalDateTime.now();
LocalDateTime nowMinTime = LocalDateTime.of(now.getYear(), now.getMonth(), now.getDayOfMonth(),
now.getHour(), now.getMinute());
// 如果上次仍未内部循环完成，则使用原来的
if(offsetFromCache.getIsNewStep()) {
offsetFromCache.setStartTime(offsetFromCache.getEndTime());
long endTime = nowMinTime.toEpochSecond(ZoneOffset.of("+8"));
offsetFromCache.setEndTime((int) endTime);
}
return offsetFromCache;
}
/**
* 生成新的完整的 偏移标识
*
* @return 新偏移
*/
private LoghubQueryCounterOffsetModel generateNewOffset() {
LoghubQueryCounterOffsetModel offsetNew = new LoghubQueryCounterOffsetModel();
LocalDateTime now = LocalDateTime.now();
LocalDateTime nowMinTime = LocalDateTime.of(now.getYear(), now.getMonth(), now.getDayOfMonth(),
now.getHour(), now.getMinute());
long startTime = nowMinTime.minusDays(1).toEpochSecond(ZoneOffset.of("+8"));
long endTime = nowMinTime.toEpochSecond(ZoneOffset.of("+8"));
offsetNew.setStartTime((int) startTime);
offsetNew.setEndTime((int) endTime);
return offsetNew;
}
/**
* 将日志返回数据 适配到数据库记录中
*
* @param logItem 日志详情
* @return db数据结构对应
*/
private BizDataStatisticsMin adaptStatisticsMinDbData(LogItem logItem) {
ArrayList<LogContent> logContents = logItem.GetLogContents();
BizDataStatisticsMin statisticsMin1 = new BizDataStatisticsMin();
for (LogContent logContent : logContents) {
switch (logContent.GetKey()) {
case "totalCountMin":
statisticsMin1.setStatisticsCount(Integer.valueOf(logContent.GetValue()));
break;
case "productCode":
statisticsMin1.setProductCode(logContent.GetValue());
break;
case "myDateTimeMinute":
String signDtMinStr = logContent.GetValue();
String[] dateTimeArr = signDtMinStr.split(" ");
String countDate = dateTimeArr[0];
String[] timeArr = dateTimeArr[1].split(":");
String countHour = timeArr[0];
String countMin = timeArr[1];
statisticsMin1.setCountDate(countDate);
statisticsMin1.setCountHour(countHour);
statisticsMin1.setCountMin(countMin);
break;
default:
break;
}
}
return statisticsMin1;
}
/**
* 重置默认值，同时提交当前 (滚动到下一个偏移点)
*/
private void rolloverOffsetAndCommit() {
resetOffsetDefaultSettings();
commitOffsetSync();
}
/**
* 提交偏移量
*
*/
private void commitOffsetSync() {
defaultOffsetQueryTaskCallback.commit();
}
} 主要实现逻辑如下:
1. 每隔一分钟进行一个查询;
2. 发生异常后，容错继续查询;
3. 对于一个新统计，默认倒推一天范围进行统计;
4. 统计时间范围间隔可设置，避免一次查询数量太大，费用太高且查询返回数量有限;
5. 对于每次小批量查询，支持分布操作，直到取完数据;
6. 小批量数据完成后，自动提交查询偏移;
7. 后续查询将基础提交的偏移进行;
8. 支持断点查询;
2. 偏移提交管理器 OffsetQueryTaskCallback
主任务中，只管进行数据统计查询，提交偏移操作由其他类进行;
/**
* 普通任务回调接口定义， 考虑到多种类型的统计任务偏移操作方式可能不一，定义一个通用型偏移接口
*
*/
public interface OffsetQueryTaskCallback {
/**
* 回调方法入口, 提交偏移
*/
public void commit();
/**
* 设置初始化绑定当前偏移（期间不得改变）
*
* @param startTime 偏移开始时间
* @param endTime 偏移结束时间
* @param offsetStart 偏移开始值（分页）
* @param limit 单次取值最大数（分页）
* @param isNewStep 是否是新的查询
* @param isSpecifyOffset 是否是指定的偏移
*/
public void initCurrentOffset(Integer startTime, Integer endTime,
Integer offsetStart, Integer limit,
Boolean isNewStep, Boolean isSpecifyOffset);
/**
* 从当前环境中获取当前偏移信息
*
* @return 偏移变量实例
*/
public LoghubQueryCounterOffsetModel getCurrentOffset();
}
import com.alibaba.fastjson.JSONObject;
import com.my.util.constants.RedisKeysConstantEnum;
import com.my.util.redis.RedisPoolUtil;
import com.my.model.LoghubQueryCounterOffsetModel;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
/**
* 默认偏移回调实现
*
*/
@Component("defaultOffsetQueryTaskCallback")
@Slf4j
public class DefaultOffsetQueryTaskCallbackImpl implements OffsetQueryTaskCallback {
@Resource
private RedisPoolUtil redisPoolUtil;
/**
* 当前偏移信息
*/
private ThreadLocal<LoghubQueryCounterOffsetModel> currentOffsetHolder = new ThreadLocal<>();
@Override
public void commit() {
if(!currentOffsetHolder.get().getIsSpecifyOffset()) {
redisPoolUtil.set(RedisKeysConstantEnum.STATISTICS_COUNTER_OFFSET_CACHE_KEY.getRedisKey(),
JSONObject.toJSONString(currentOffsetHolder.get()));
}
}
@Override
public void initCurrentOffset(Integer startTime, Integer endTime,
Integer offsetStart, Integer limit,
Boolean isNewStep, Boolean isSpecifyOffset) {
LoghubQueryCounterOffsetModel currentOffset = new LoghubQueryCounterOffsetModel();
currentOffset.setStartTime(startTime);
currentOffset.setEndTime(endTime);
currentOffset.setOffsetStart(offsetStart);
currentOffset.setIsNewStep(isNewStep);
currentOffset.setIsSpecifyOffset(isSpecifyOffset);
currentOffsetHolder.set(currentOffset);
}
@Override
public LoghubQueryCounterOffsetModel getCurrentOffset() {
return currentOffsetHolder.get();
}
/**
* 从缓存中获取当前偏移信息
*
* @return 缓存偏移或者 null
*/
public LoghubQueryCounterOffsetModel getCurrentOffsetFromCache() {
String offsetCacheValue = redisPoolUtil.get(RedisKeysConstantEnum.STATISTICS_COUNTER_OFFSET_CACHE_KEY.getRedisKey());
if (StringUtils.isBlank(offsetCacheValue)) {
return null;
}
return JSONObject.parseObject(offsetCacheValue, LoghubQueryCounterOffsetModel.class);
}
public Integer getStartTime() {
return currentOffsetHolder.get().getStartTime();
}
public void setStartTime(Integer startTime) {
currentOffsetHolder.get().setStartTime(startTime);
}
public Integer getEndTime() {
return currentOffsetHolder.get().getEndTime();
}
public void setEndTime(Integer endTime) {
currentOffsetHolder.get().setEndTime(endTime);
}
public Integer getOffsetStart() {
return currentOffsetHolder.get().getOffsetStart();
}
public void setOffsetStart(Integer offsetStart) {
currentOffsetHolder.get().setOffsetStart(offsetStart);
}
public Integer getLimit() {
return currentOffsetHolder.get().getLimit();
}
public void setLimit(Integer limit) {
currentOffsetHolder.get().setLimit(limit);
}
public Boolean getIsNewStep() {
return currentOffsetHolder.get().getIsNewStep();
}
public void setIsNewStep(Boolean isNewStep) {
currentOffsetHolder.get().setIsNewStep(isNewStep);
}
}
/**
* loghub 查询偏移量 数据容器
*
*/
@Data
public class LoghubQueryCounterOffsetModel implements Serializable {
private static final long serialVersionUID = -3749552331349228045L;
/**
* 开始时间
*/
private Integer startTime;
/**
* 结束时间
*/
private Integer endTime;
/**
* 起始偏移
*/
private Integer offsetStart = 0;
/**
* 每次查询的 条数限制, 都需要进行设置后才可用, 否则查无数据
*/
private Integer limit = 0;
/**
* 是否新的偏移循环，如未完成，应继续子循环 limit
*
* true: 是, offsetStart,limit 失效, false: 否, 需借助 offsetStart,limit 进行limit相加
*/
private Boolean isNewStep = true;
/**
* 是否是手动指定的偏移，如果是说明是在手动被数据，偏移量将不会被更新
*
* 此变量是瞬时值，将不会被持久化到偏移标识中
*/
private transient Boolean isSpecifyOffset;
} 3. 批量更新统计结果数据库的实现
因每次统计的数据量是不确定的，因尽可能早的提交一次统计结果，防止一次提交太多，或者 机器故障时所有统计白费，所以需要分小事务进行。
@Service
public class StatisticsServiceImpl implements StatisticsService {
/**
* 批量更新统计分钟级数据 （事务型提交）
*
* @param statisticsMinList 新统计数据
* @return 影响行数
*/
@Transactional(isolation = Isolation.READ_COMMITTED, rollbackFor = Throwable.class)
public Integer batchUpsertPremiumStatistics(List<BizProposalPolicyStatisticsMin> statisticsMinList,
OffsetQueryTaskCallback callback) {
AtomicInteger updateCount = new AtomicInteger(0);
statisticsMinList.forEach(item -> {
int affectNum = 0;
BizProposalPolicyStatisticsMin oldStatistics = bizProposalPolicyStasticsMinMapper.selectOneByCond(item);
if (oldStatistics == null) {
item.setEtlVersion(item.getEtlVersion() + ":0");
affectNum = bizProposalPolicyStasticsMinMapper.insert(item);
} else {
oldStatistics.setStatisticsCount(oldStatistics.getStatisticsCount() + item.getStatisticsCount());
String versionFull = versionKeeperFilter(oldStatistics.getEtlVersion(), item.getEtlVersion());
oldStatistics.setEtlVersion(versionFull + ":" + oldStatistics.getStatisticsCount());
// todo: 优化更新版本号问题
affectNum = bizProposalPolicyStasticsMinMapper.updateByPrimaryKey(oldStatistics);
}
updateCount.addAndGet(affectNum);
});
callback.commit();
return updateCount.get();
}
/**
* 版本号过滤器（组装版本信息）
*
* @param oldVersion 老版本信息
* @param currentVersion 当前版本号
* @return 可用的版本信息
*/
private String versionKeeperFilter(String oldVersion, String currentVersion) {
String versionFull = oldVersion + "," + currentVersion;
if (versionFull.length() >= 500) {
// 从150以后，第一版本号开始保留
versionFull = versionFull.substring(versionFull.indexOf(',', 150));
}
return versionFull;
}
} 4. 你需要一个启动任务的地方
/**
* 启动时运行的任务调度服务
*
*/
@Service
@Slf4j
public class TaskAutoRunScheduleService {
@Resource
private MinuteBizDataCounterTask minuteBizDataCounterTask;
@PostConstruct
public void bizDataAutoRun() {
log.info("============= bizDataAutoRun start =================");
ExecutorService executorService = Executors.newSingleThreadExecutor(new NamedThreadFactory("Biz-data-counter-%d"));
executorService.submit(minuteBizDataCounterTask);
}
} 5. 将每分钟的数据从db查询出来展示到页面
以上将数据统计后以分钟级汇总到数据，接下来，监控页面就只需从db中进行简单聚合就可以了，咱们就不费那精力去展示了。
6. 待完善的地方
1. 集群环境的任务运行将会出问题，解决办法是：加一个分布式锁即可。 你可以的！
2. 针对重试执行统计问题，还得再考虑考虑。（幂等性）
唠叨: 踩坑不一定是坏事！
不要害怕今日的苦，你要相信明天，更苦！
![虚拟机---kafka的安装[图]](data:image/gif;base64,R0lGODlhAQABAIAAAP///wAAACwAAAAAAQABAAACAkQBADs=)