使用SpringBoot+InfluxDB实现高效数据存储与查询

# InfluxDB 独立配置
influxdb:
  url: http://192.168.1.1xx:28086/  # InfluxDB 服务器地址
  token: _7FZlXGJJcd8Ayox-F-hVBDdXb_a5SI3530x1DdFKZfQ65uOhnpQciJWHpd7ULhpAOcgj5oV2JsR-Xf0qTtAxg==
  org: xxx     # 组织名称
  bucket: xxx  # 存储桶名称
  # InfluxDB 客户端日志级别
  # ERROR: 仅记录错误日志
  # WARN: 记录警告和错误日志
  # INFO: 记录普通信息、警告和错误日志
  # DEBUG: 记录调试级别的详细日志
  # BODY: 记录完整的 HTTP 请求和响应主体
  # TRACE: 记录极其详细的跟踪日志
  # ALL: 记录所有日志级别（视客户端而定）
  logLevel: BODY

import com.influxdb.client.InfluxDBClient;
import com.influxdb.client.InfluxDBClientFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
@Configuration
public class InfluxDBConfig {
 
    @Value("${influxdb.url}")
    private String url;
 
    @Value("${influxdb.token}")
    private String token;
 
    @Value("${influxdb.org}")
    private String org;
 
    @Value("${influxdb.bucket}")
    private String bucket;
 
    @Bean
    public InfluxDBClient influxDBClient() {
        return InfluxDBClientFactory.create(url, token.toCharArray(), org, bucket);
    }
}

public void writeSingleData(TemperatureDTO temperatureDTO) {
    WriteApiBlocking writeApi = influxDBClient.getWriteApiBlocking();
    
    Point point = Point.measurement("temperature")
            .addTag("location", temperatureDTO.getLocation()) // 添加标签（索引）
            .addField("value", temperatureDTO.getValue()) // 添加字段（数据）
            .time(Instant.now(), WritePrecision.NS); // 记录当前时间戳
 
    writeApi.writePoint(point);
}

public void writeBatchData(List<TemperatureDTO> temperatureDTOs) {
    WriteApi writeApi = influxDBClient.makeWriteApi(); // 获取异步 API
 
    List<Point> points = temperatureDTOs.stream()
            .map(dto -> Point.measurement("temperature")
                    .addTag("location", dto.getLocation())
                    .addField("value", dto.getValue())
                    .time(Instant.now(), WritePrecision.NS))
            .collect(Collectors.toList());
 
    CompletableFuture<Void> future = CompletableFuture.runAsync(() -> writeApi.writePoints(points));
    future.whenComplete((result, error) -> {
        if (error != null) {
            System.err.println("🔥 写入失败：" + error.getMessage());
        } else {
            writeApi.close();  // 关闭 API 避免资源泄露
            log.info("✅ 批量数据写入成功（异步）");
        }
    });
}

public List<TemperatureVO> queryTemperatureData() {
    String query = "from(bucket: \"test\") |> range(start: -1h)";
    QueryApi queryApi = influxDBClient.getQueryApi();
 
    return queryApi.query(query)
            .stream()
            .flatMap(fluxTable -> fluxTable.getRecords().stream()) // 遍历 FluxTable
            .map(record -> {
                TemperatureVO vo = new TemperatureVO();
                vo.setLocation((String) record.getValueByKey("location")); // 获取标签信息
                Object valueObj = record.getValueByKey("_value");
                vo.setValue(valueObj != null ? ((Number) valueObj).doubleValue() : 0.0);
                vo.setTimestamp(record.getTime().toString());
                return vo;
            })
            .collect(Collectors.toList());
}

@RestController
@RequestMapping("/api/influxdb")
public class InfluxDBController {
 
    @Autowired
    private TestService influxDBService;
 
    @PostMapping("/write")
    public String writeData(@RequestBody TemperatureDTO temperatureDTO) {
        influxDBService.writeSingleData(temperatureDTO);
        return "✅ 单条数据写入成功！";
    }
 
    @PostMapping("/write-batch")
    public String writeBatchData() {
        List<TemperatureDTO> data = generateTestData(10000);
        influxDBService.writeBatchData(data);
        return "✅ 10,000 条数据成功写入！";
    }
 
    @GetMapping("/query")
    public List<TemperatureVO> queryTemperatureData() {
        return influxDBService.queryTemperatureData();
    }
 
    private List<TemperatureDTO> generateTestData(int count) {
        List<TemperatureDTO> dataList = new ArrayList<>();
        Random random = new Random();
        for (int i = 0; i < count; i++) {
            TemperatureDTO dto = new TemperatureDTO();
            dto.setLocation("office-" + (random.nextInt(1000) + 1));
            dto.setValue(15 + (random.nextDouble() * 10));
            dataList.add(dto);
        }
        return dataList;
    }
}

mvn spring-boot:run 

{
  "location": "office-1",
  "value": 22.5
}