实习经历中的一次项目优化（物联网项目）

104 阅读 0 评论 69 点赞

我是靠谱客的博主高贵酒窝，最近开发中收集的这篇文章主要介绍实习经历中的一次项目优化（物联网项目），觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

在实习过程中，接到一个需求，要查询某一个时间段中InfluxDB的时序数据，并且要求必须按照时间有序进行排列，更重要的是快。然后将数据以CSV文件导出

传统做法：

influxDB中的数据本身就是按照时间排序的，所以可以查询该时间段的时序数据；为了提高查询效率，influxDB可以开启Gzip压缩机制，可以减少1/3的耗时。但是经过测试，如果要查询某一台设备两年内的时序数据，数据量至少为百万起步，甚至达到千万；测试显示直接查询该时间段数据需要耗时2-3小时。

优化方案（采用线程池+CountDownLanch实现）：

这个业务场景下可以基于时间段进行拆分，划分为多个子任务，任务异步执行提高效率，然后用CountDownLanch计数器计数，保证所有数据查询完成后进行数据导出。为保证数据有序，在时间段分割过程中，我给每个时间段标上索引位，保证后面数据是有序的，代码如下：

线程池的配置：

package com.ren.mqtt.config;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
/**
* @ClassName: ThreadPoolConfig
* @Description: TODO
* @Author: RZY
* @DATE: 2022/8/18 12:31
* @Version: v1.0
*/
@Configuration
public class ThreadPoolConfig {
//核心线程数
@Value("${ThreadPool.corePoolSize}")
private int corePoolSize;
//最大线程数
@Value("${ThreadPool.maxPoolSize}")
private int maxPoolSize;
//工作线程活跃时间
@Value("${ThreadPool.keepAliveTime}")
private int keepAliveTime;
//阻塞队列容量
@Value("${ThreadPool.BlockingQueueCapacity}")
private int BlockingQueueCapacity;
@Bean("influxDBThreadPool")
public ThreadPoolTaskExecutor threadPoolTaskExecutor(ThreadFactory threadFactory) {
ThreadPoolTaskExecutor threadPool = new ThreadPoolTaskExecutor();
threadPool.setCorePoolSize(corePoolSize);
threadPool.setMaxPoolSize(maxPoolSize);
threadPool.setKeepAliveSeconds(keepAliveTime);
threadPool.setQueueCapacity(BlockingQueueCapacity);
threadPool.setThreadFactory(threadFactory);
// 线程池对拒绝任务的处理策略
// CallerRunsPolicy：由调用线程（提交任务的线程）处理该任务
threadPool.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
// 初始化
threadPool.initialize();
return threadPool;
}
@Bean
public ThreadFactory threadFactory() {
//自定义线程工厂
return new ThreadFactoryBuilder().setNameFormat("influxDB-thread-pool-").build();
}
}

数据下载api：

package com.ren.mqtt.controller;
import com.alibaba.excel.EasyExcel;
import com.alibaba.excel.util.ListUtils;
import com.influxdb.client.InfluxDBClient;
import com.influxdb.client.QueryApi;
import com.ren.mqtt.utils.MqttUtils;
import com.ren.mqtt.utils.StringFilterUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.net.URLEncoder;
import java.time.Duration;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.ZoneOffset;
import java.util.*;
import java.util.concurrent.*;
/**
* @ClassName: EasyExcelController
* @Description: TODO
* @Author: RZY
* @DATE: 2022/6/27 16:46
* @Version: v1.0
*/
@RestController
@Slf4j
public class EasyExcelController {
@Resource
MqttUtils mqttUtils;
@Resource
InfluxDBClient influxDBClient;
@Resource
ThreadPoolExecutor influxDBThreadPool;
/**
* 下载influxDB中的时序数据，并以CSV格式导出
* @param request
* @param response
* @throws Exception
*/
@GetMapping("")
public void upload(HttpServletRequest request, HttpServletResponse response) throws Exception {
QueryApi queryApi = influxDBClient.getQueryApi();
long start = 1653970085L;
long end = 1665976485L;
Long deviceId = 1394180959449542657L;
LocalDateTime startTime = LocalDateTime.ofEpochSecond(start, 0, ZoneOffset.ofHours(8));
LocalDateTime endTime = LocalDateTime.ofEpochSecond(end, 0, ZoneOffset.ofHours(8));
//是否采用线程池
boolean flag = false;
//判断时间段之差是否大于3天，小于3天则不需要采用线程池，否则采用线程池提高效率
if(Duration.between(startTime, endTime).toHours() > 72) flag = true;
// 获取开始时间和结束时间的小时差
Map<Long, Long> dateMap = new LinkedHashMap<>();
cutDateTime(startTime, endTime, flag ? 16 : 1, dateMap);
//保证线程安全采用ConcurrentHashMap
Map<Integer, List<List<Object>>> data = new ConcurrentHashMap<>();
if(flag) {
//countDownLatch实现异步转同步
CountDownLatch countDownLatch = new CountDownLatch(16);
//线程池任务异步执行
final int[] index = {0};
dateMap.forEach((key, value) -> {
influxDBThreadPool.execute(() -> {
queryFluxAsync(queryApi, countDownLatch, deviceId, key, value, index[0]++, data);
});
});
//唤醒主线程
countDownLatch.await(30, TimeUnit.SECONDS);
} else queryFlux(queryApi, deviceId, startTime.toEpochSecond(ZoneOffset.of("+8"), endTime.toEpochSecond(ZoneOffset.of("+8"), data);
//最终时序数据整合
List<List<Object>> res = new ArrayList<>();
for (int i = 0; i < 10; i++) {
res.addAll(data.get(i));
}
// 查到数据，准备导出csv数据
log.info("查询到数据，准备导出excel");
response.setContentType("application/vnd.openxmlformats-officedocument.spreadsheetml.sheet");
response.setCharacterEncoding("utf-8");
// 这里URLEncoder.encode可以防止中文乱码 当然和easyexcel没有关系
String fileName = URLEncoder.encode("测试", "UTF-8").replaceAll("\+", "%20");
response.setHeader("Content-disposition", "attachment;filename*=utf-8''" + fileName + ".xlsx");
EasyExcel.write(response.getOutputStream()).sheet("模板").doWrite(res);
}
private void queryFluxAsync(QueryApi queryApi, CountDownLatch countDownLatch, Long deviceId, Long start, Long end, int index, Map<Integer, List<List<Object>>> map) {
List<List<Object>> data = ListUtils.newArrayList();
//查询该时间段的时序数据
queryApi.query(getFlux(deviceId, start, end), (k, v) -> {
String queryValue = Objects.requireNonNull(v.getValue()).toString().replaceAll("rn", ",").replace(" ","");
if(StringFilterUtils.isMessyCode(queryValue)) {
//一行数据
ArrayList<Object> queryData = ListUtils.newArrayList();
queryData.add(Objects.requireNonNull(v.getTime()).atZone(ZoneId.systemDefault()).toString());
//String[] split = queryValue.split(",");
//queryData.addAll(Arrays.asList(split));
//封装查询的时序数据
queryData.add(queryValue);
data.add(queryData);
}
}, error -> {
//计数器减一，查询失败前提下也要释放，避免一直阻塞
countDownLatch.countDown();
log.info("查询失败");
}, () -> {
map.put(index, data);
log.info("<---------->data:" + data.size());
//计数器减一，查询任务执行成功的前提下
countDownLatch.countDown();
});
}
private void queryFlux(QueryApi queryApi, Long deviceId, Long start, Long end, Map<Integer, List<List<Object>>> map) {
List<List<Object>> data = ListUtils.newArrayList();
//查询该时间段的时序数据
queryApi.query(getFlux(deviceId, start, end), (k, v) -> {
String queryValue = Objects.requireNonNull(v.getValue()).toString().replaceAll("rn", ",").replace(" ","");
if(StringFilterUtils.isMessyCode(queryValue)) {
//一行数据
ArrayList<Object> queryData = ListUtils.newArrayList();
queryData.add(Objects.requireNonNull(v.getTime()).atZone(ZoneId.systemDefault()).toString());
//String[] split = queryValue.split(",");
//queryData.addAll(Arrays.asList(split));
//封装查询的时序数据
queryData.add(queryValue);
data.add(queryData);
}
}, error -> {
log.info("查询失败");
}, () -> {
map.put(0, data);
});
}
private String getFlux(Long deviceId, Long start, Long end) {
return String.format("from(bucket: "gsdzsz") |> range(start: %s, stop: %s) |> filter(fn: (r) => r._measurement == "%s" and r._value !="" and r._value != "mm") |> keep(columns: ["_time", "_value"])", start, end, deviceId);
}
private void cutDateTime(LocalDateTime start, LocalDateTime end, int count, Map<Long, Long> map) {
// 获取开始时间和结束时间的小时差
long hours = Duration.between(start, end).toHours() / count;
for(int i=0;i<count - 1;i++) {
map.put(start.plusHours(hours * i).toEpochSecond(ZoneOffset.of("+8")),
start.plusHours(hours * (i + 1)).toEpochSecond(ZoneOffset.of("+8")));
}
//封装时间段
map.put(start.plusHours(hours * 9).toEpochSecond(ZoneOffset.of("+8")),
end.toEpochSecond(ZoneOffset.of("+8")));
}
}

注意：是否采用线程池，还需要判断一下时间段的大小，如果是查询3天内数据，其实没必要采用线程池，直接让主线程执行就可以，因为此时数据量只有1万+，查询只需几秒，就没必要用线程池加速了。另外CountDownLanch的计数器减一的代码必须放在queryApi.query()方法的异步回调函数中执行，因此这个查询是异步执行的，直接在任务中减一会导致数据还没有查出，计数器就变为0，导出空数据，这是一个bug（尤为注意）

总结：采用了线程池加速后，百万级数据查询只需要几分钟（相较于之前的2-3小时），而且这是本机测试结果，如果在服务器上测试，效果更显著。