diff --git a/pom.xml b/pom.xml index 8903da4..662c59e 100644 --- a/pom.xml +++ b/pom.xml @@ -77,7 +77,7 @@ com.alibaba fastjson - 1.2.15 + 1.2.83 com.google.code.gson @@ -134,6 +134,14 @@ + + org.apache.maven.plugins + maven-compiler-plugin + + 8 + 8 + + diff --git a/src/main/java/com/topsail/influxdb/config/InfluxDBConfig.java b/src/main/java/com/topsail/influxdb/config/InfluxDBConfig.java new file mode 100644 index 0000000..04e4c76 --- /dev/null +++ b/src/main/java/com/topsail/influxdb/config/InfluxDBConfig.java @@ -0,0 +1,409 @@ +package com.topsail.influxdb.config; + +import com.influxdb.client.InfluxDBClient; +import com.influxdb.client.InfluxDBClientFactory; +import com.influxdb.client.InfluxDBClientOptions; +import com.influxdb.client.WriteApi; +import com.influxdb.client.WriteOptions; +import com.influxdb.client.write.events.WriteErrorEvent; +import com.influxdb.client.write.events.WriteSuccessEvent; +import okhttp3.OkHttpClient; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +import javax.annotation.PreDestroy; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; + +/** + * InfluxDB配置类 + * 负责管理InfluxDB客户端和WriteApi的生命周期 + */ +@Configuration +public class InfluxDBConfig { + + private static final Logger LOG = LoggerFactory.getLogger(InfluxDBConfig.class); + + @Value("${shengdilan.influxdb.url}") + private String url; + + @Value("${shengdilan.influxdb.token}") + private String token; + + @Value("${shengdilan.influxdb.org}") + private String org; + + private InfluxDBClient influxDBClient; + + /** 标记InfluxDB连接是否可用 */ + private final AtomicBoolean connectionHealthy = new AtomicBoolean(true); + /** 连续写入失败计数 */ + private final AtomicLong consecutiveFailures = new AtomicLong(0); + /** 连接不可用标志 - 当连续失败超过阈值时置为true */ + private static final long FAILURE_THRESHOLD = 5; + + /** 定时健康检查调度器 */ + private ScheduledExecutorService healthCheckScheduler; + /** 健康检查最小间隔（秒） */ + private static final int HEALTH_CHECK_MIN_INTERVAL_SECONDS = 3; + /** 健康检查最大间隔（秒） */ + private static final int HEALTH_CHECK_MAX_INTERVAL_SECONDS = 60; + /** 当前健康检查间隔（秒），用于指数退避 */ + private final AtomicLong currentHealthCheckInterval = new AtomicLong(HEALTH_CHECK_MIN_INTERVAL_SECONDS); + /** 健康检查连续失败计数 */ + private final AtomicLong healthCheckFailCount = new AtomicLong(0); + /** 健康检查连续失败多少次后重建客户端 */ + private static final long HEALTH_CHECK_RECREATE_THRESHOLD = 3; + + /** 当前使用的WriteApi引用（重建客户端时同步更新） */ + private volatile WriteApi currentWriteApi; + + /** 连接恢复回调监听器列表（线程安全） */ + private final List recoveryCallbacks = new CopyOnWriteArrayList<>(); + + /** + * 创建InfluxDB客户端Bean + * 配置优化的超时参数 - 针对网络不稳定场景优化 + * @return InfluxDBClient实例 + */ + @Bean + public InfluxDBClient influxDBClient() { + // 配置 OkHttpClient 的超时参数 + OkHttpClient.Builder okHttpClientBuilder = new OkHttpClient.Builder() + .connectTimeout(10, TimeUnit.SECONDS) // 连接超时：10秒（缩短以快速检测不可达） + .readTimeout(30, TimeUnit.SECONDS) // 读取超时：30秒 + .writeTimeout(30, TimeUnit.SECONDS); // 写入超时：30秒 + + // 使用自定义的 OkHttpClient Builder 创建 InfluxDB 客户端 + InfluxDBClientOptions options = InfluxDBClientOptions.builder() + .url(url) + .authenticateToken(token.toCharArray()) + .org(org) + .okHttpClient(okHttpClientBuilder) + .build(); + + this.influxDBClient = InfluxDBClientFactory.create(options); + LOG.info("InfluxDB客户端已创建，连接地址: {}", url); + return this.influxDBClient; + } + + /** + * 创建WriteApi Bean + * 配置优化的批量写入选项 - 针对高吞吐量场景优化 + * 添加写入事件监听器，捕获异步写入错误并将失败数据备份到MQ + * @param client InfluxDB客户端 + * @return WriteApi实例 + */ + @Bean + public WriteApi writeApi(InfluxDBClient client) { + WriteOptions writeOptions = WriteOptions.builder() + .batchSize(2) //2 批量大小：增加到5000，提高批量写入效率 + .flushInterval(2000) // 2000刷新间隔：2秒，平衡实时性和性能 + .bufferLimit(10000) // 100000缓冲区限制：增加到10万，避免背压警告 + .retryInterval(3000) // 3000重试间隔：3秒 + .build(); + + WriteApi writeApi = client.getWriteApi(writeOptions); + this.currentWriteApi = writeApi; + + // 注册写入成功事件监听器 - 写入成功时检测状态转换并触发恢复回调 + writeApi.listenEvents(WriteSuccessEvent.class, event -> { + // 检测从不可用到可用的状态转换，仅在转换时触发恢复回调 + boolean wasUnhealthy = !connectionHealthy.getAndSet(true); + if (wasUnhealthy) { + consecutiveFailures.set(0); + healthCheckFailCount.set(0); + currentHealthCheckInterval.set(HEALTH_CHECK_MIN_INTERVAL_SECONDS); + LOG.info("InfluxDB连接已通过异步写入成功恢复正常，触发恢复回调"); + fireRecoveryCallbacks(); + } else { + consecutiveFailures.set(0); + } + }); + + // 注册写入错误事件监听器 - 捕获异步写入失败 + writeApi.listenEvents(WriteErrorEvent.class, event -> { + long failures = consecutiveFailures.incrementAndGet(); + Throwable throwable = event.getThrowable(); + String errorMsg = throwable != null ? throwable.getMessage() : "未知错误"; + + if (failures >= FAILURE_THRESHOLD) { + connectionHealthy.set(false); + LOG.error("InfluxDB写入连续失败{}次，标记连接为不可用状态。错误: {}", failures, errorMsg); + } else { + LOG.warn("InfluxDB异步写入失败(第{}次): {}", failures, errorMsg); + } + }); + + LOG.info("WriteApi已创建，已注册写入事件监听器"); + + // 启动定时健康检查 + startHealthCheck(); + + return writeApi; + } + + /** + * 注册连接恢复回调监听器 + * 当InfluxDB连接从不可用恢复为可用时，会依次调用所有已注册的回调 + * @param callback 恢复回调 + */ + public void addRecoveryCallback(Runnable callback) { + if (callback != null) { + recoveryCallbacks.add(callback); + LOG.info("已注册InfluxDB连接恢复回调，当前回调数量: {}", recoveryCallbacks.size()); + } + } + + /** + * 触发所有连接恢复回调 + */ + private void fireRecoveryCallbacks() { + if (recoveryCallbacks.isEmpty()) { + return; + } + LOG.info("开始触发InfluxDB连接恢复回调，共{}个", recoveryCallbacks.size()); + for (Runnable callback : recoveryCallbacks) { + try { + callback.run(); + } catch (Exception e) { + LOG.error("执行连接恢复回调时发生异常: {}", e.getMessage(), e); + } + } + } + + /** + * 启动定时健康检查（带指数退避策略） + * 当连接不可用时，从3秒开始，每次失败后间隔翻倍，最大60秒 + * 连接恢复后重置间隔 + * 健康检查连续失败超过阈值时，自动重建InfluxDB客户端连接 + */ + private void startHealthCheck() { + healthCheckScheduler = Executors.newSingleThreadScheduledExecutor(r -> { + Thread t = new Thread(r, "influxdb-health-check"); + t.setDaemon(true); + return t; + }); + + // 使用递归调度实现动态间隔 + scheduleNextHealthCheck(); + + LOG.info("InfluxDB定时健康检查已启动，初始间隔: {}秒，最大间隔: {}秒，重建阈值: {}次", + HEALTH_CHECK_MIN_INTERVAL_SECONDS, HEALTH_CHECK_MAX_INTERVAL_SECONDS, HEALTH_CHECK_RECREATE_THRESHOLD); + } + + /** + * 调度下一次健康检查（实现指数退避 + 客户端自动重建） + */ + private void scheduleNextHealthCheck() { + long interval = currentHealthCheckInterval.get(); + healthCheckScheduler.schedule(() -> { + // 仅在连接不可用时才进行健康检查 + if (!connectionHealthy.get()) { + try { + boolean pingResult = influxDBClient.health().getStatus().equals("pass"); + if (pingResult) { + long failCount = healthCheckFailCount.get(); + LOG.info("InfluxDB健康检查通过，恢复连接状态（上次检查间隔: {}秒，累计失败: {}次）", interval, failCount); + healthCheckFailCount.set(0); + connectionHealthy.set(true); + consecutiveFailures.set(0); + // 重置退避间隔 + currentHealthCheckInterval.set(HEALTH_CHECK_MIN_INTERVAL_SECONDS); + // 触发恢复回调 + fireRecoveryCallbacks(); + } else { + long failCount = healthCheckFailCount.incrementAndGet(); + LOG.warn("InfluxDB健康检查未通过（连续失败{}次），下次检查间隔将增加", failCount); + checkAndRecreateClient(failCount); + increaseHealthCheckInterval(); + } + } catch (Exception e) { + long failCount = healthCheckFailCount.incrementAndGet(); + LOG.warn("InfluxDB健康检查异常（连续失败{}次）: {}，下次检查间隔将增加", failCount, e.getMessage()); + checkAndRecreateClient(failCount); + increaseHealthCheckInterval(); + } + } else { + // 连接健康，重置间隔和失败计数 + healthCheckFailCount.set(0); + currentHealthCheckInterval.set(HEALTH_CHECK_MIN_INTERVAL_SECONDS); + } + // 调度下一次检查 + scheduleNextHealthCheck(); + }, interval, TimeUnit.SECONDS); + } + + /** + * 检查健康检查失败次数是否达到阈值，达到则重建InfluxDB客户端和WriteApi + * @param failCount 当前连续失败次数 + */ + private synchronized void checkAndRecreateClient(long failCount) { + if (failCount >= HEALTH_CHECK_RECREATE_THRESHOLD && failCount % HEALTH_CHECK_RECREATE_THRESHOLD == 0) { + LOG.warn("健康检查连续失败{}次达到阈值，开始重建InfluxDB客户端连接...", failCount); + try { + recreateInfluxDBClient(); + LOG.info("InfluxDB客户端重建完成"); + } catch (Exception e) { + LOG.error("InfluxDB客户端重建失败: {}", e.getMessage(), e); + } + } + } + + /** + * 重建InfluxDB客户端和WriteApi + * 关闭旧的连接池和WriteApi，创建全新的实例，解决连接池中的失效连接导致健康检查持续失败的问题 + */ + private synchronized void recreateInfluxDBClient() { + try { + // 1. 关闭旧的WriteApi + if (currentWriteApi != null) { + try { + currentWriteApi.close(); + LOG.info("旧WriteApi已关闭"); + } catch (Exception e) { + LOG.warn("关闭旧WriteApi时发生异常: {}", e.getMessage()); + } + } + // 2. 关闭旧的InfluxDBClient + if (influxDBClient != null) { + try { + influxDBClient.close(); + LOG.info("旧InfluxDB客户端已关闭"); + } catch (Exception e) { + LOG.warn("关闭旧InfluxDB客户端时发生异常: {}", e.getMessage()); + } + } + // 3. 创建全新的InfluxDBClient + OkHttpClient.Builder okHttpClientBuilder = new OkHttpClient.Builder() + .connectTimeout(10, TimeUnit.SECONDS) + .readTimeout(30, TimeUnit.SECONDS) + .writeTimeout(30, TimeUnit.SECONDS); + InfluxDBClientOptions options = InfluxDBClientOptions.builder() + .url(url) + .authenticateToken(token.toCharArray()) + .org(org) + .okHttpClient(okHttpClientBuilder) + .build(); + this.influxDBClient = InfluxDBClientFactory.create(options); + LOG.info("新InfluxDB客户端已创建，连接地址: {}", url); + // 4. 创建全新的WriteApi并注册事件监听器 + WriteOptions writeOptions = WriteOptions.builder() + .batchSize(2) // 批量大小：与Bean配置保持一致 + .flushInterval(2000) + .bufferLimit(10000) + .retryInterval(3000) + .build(); + WriteApi newWriteApi = influxDBClient.getWriteApi(writeOptions); + this.currentWriteApi = newWriteApi; + // 重新注册写入成功事件监听器 + newWriteApi.listenEvents(WriteSuccessEvent.class, event -> { + boolean wasUnhealthy = !connectionHealthy.getAndSet(true); + if (wasUnhealthy) { + consecutiveFailures.set(0); + healthCheckFailCount.set(0); + currentHealthCheckInterval.set(HEALTH_CHECK_MIN_INTERVAL_SECONDS); + LOG.info("InfluxDB连接已通过异步写入成功恢复正常（重建后），触发恢复回调"); + fireRecoveryCallbacks(); + } else { + consecutiveFailures.set(0); + } + }); + // 重新注册写入错误事件监听器 + newWriteApi.listenEvents(WriteErrorEvent.class, event -> { + long failures = consecutiveFailures.incrementAndGet(); + Throwable throwable = event.getThrowable(); + String errorMsg = throwable != null ? throwable.getMessage() : "未知错误"; + if (failures >= FAILURE_THRESHOLD) { + connectionHealthy.set(false); + LOG.error("InfluxDB写入连续失败{}次（重建后），标记连接为不可用状态。错误: {}", failures, errorMsg); + } else { + LOG.warn("InfluxDB异步写入失败(第{}次，重建后): {}", failures, errorMsg); + } + }); + LOG.info("新WriteApi已创建并注册事件监听器"); + } catch (Exception e) { + LOG.error("重建InfluxDB客户端过程中发生异常: {}", e.getMessage(), e); + } + } + + /** + * 增加健康检查间隔（指数退避，上限为MAX） + */ + private void increaseHealthCheckInterval() { + long current = currentHealthCheckInterval.get(); + long next = Math.min(current * 2, HEALTH_CHECK_MAX_INTERVAL_SECONDS); + currentHealthCheckInterval.set(next); + LOG.debug("健康检查间隔从{}秒增加到{}秒", current, next); + } + + /** + * 检查InfluxDB连接是否健康 + * @return true表示连接正常，false表示连接不可用 + */ + public boolean isConnectionHealthy() { + return connectionHealthy.get(); + } + + /** + * 立即标记连接为不健康状态 + * 当同步写入遇到超时/连接异常时调用，避免其他线程继续尝试无效写入 + */ + public void markConnectionUnhealthy() { + if (connectionHealthy.getAndSet(false)) { + LOG.warn("InfluxDB连接已被同步写入异常标记为不可用状态，后续数据将直接发送到MQ备份"); + } + } + + /** + * 获取当前可用的WriteApi实例 + * 当客户端被重建后，返回的是新的WriteApi实例 + * @return 当前WriteApi实例 + */ + public WriteApi getCurrentWriteApi() { + return currentWriteApi; + } + + /** + * 获取组织名称 + * @return 组织名称 + */ + public String getOrg() { + return org; + } + + /** + * 获取InfluxDB连接URL + * @return 连接URL + */ + public String getUrl() { + return url; + } + + /** + * 应用关闭时清理资源 + */ + @PreDestroy + public void destroy() { + // 关闭健康检查调度器 + if (healthCheckScheduler != null) { + healthCheckScheduler.shutdown(); + } + if (influxDBClient != null) { + try { + influxDBClient.close(); + } catch (Exception e) { + // 记录日志但不抛出异常，避免影响关闭流程 + LOG.warn("关闭InfluxDB客户端时发生异常: {}", e.getMessage()); + } + } + } +} diff --git a/src/main/java/com/topsail/influxdb/rabbitmq/AmqpListener.java b/src/main/java/com/topsail/influxdb/rabbitmq/AmqpListener.java index 2c1b656..1e59992 100644 --- a/src/main/java/com/topsail/influxdb/rabbitmq/AmqpListener.java +++ b/src/main/java/com/topsail/influxdb/rabbitmq/AmqpListener.java @@ -1,6 +1,7 @@ package com.topsail.influxdb.rabbitmq; import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson.JSONException; import com.rabbitmq.client.Channel; import com.topsail.influxdb.entity.DeviceLogData; import com.topsail.influxdb.pojo.History; @@ -17,6 +18,9 @@ import org.springframework.messaging.handler.annotation.Payload; import org.springframework.stereotype.Component; import org.springframework.util.StringUtils; +import java.util.HashSet; +import java.util.Set; + @Component public class AmqpListener { public static final Logger LOG = LoggerFactory.getLogger(AmqpListener.class); @@ -36,24 +40,112 @@ public class AmqpListener { * @param channel * @throws Exception */ - - @RabbitListener(queues = "shengdilandevicedataall") +// @RabbitListener(queues = "shengdilandevicedataall") public void deviceDataMqListener(@Payload String message, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Channel channel) throws Exception { if (StringUtils.isEmpty(message)) { channel.basicAck(deliveryTag, false); return; } + try { History history = JSON.parseObject(message, History.class); deviceDataService.saveDeviceDataToInfluxdb(history); - LOG.info("saveDeviceDataToInfluxdb OK:" + history.getImei()); + + // 成功写入后确认消息 + channel.basicAck(deliveryTag, false); + } catch (JSONException je) { + // ⚠️ 毒丸保护：JSON解析异常不会通过重试恢复，直接丢弃消息并记录日志 + LOG.error("JSON解析异常（不可恢复），消息将被丢弃: {}", je.getMessage(), je); + LOG.error("被丢弃的消息内容: {}", message); channel.basicAck(deliveryTag, false); } catch (Exception e) { - channel.basicNack(deliveryTag, false, true); - LOG.info("saveDeviceDataToInfluxdb Error:(" + e + ")-" + message); + LOG.error("处理设备数据时发生异常: {}", e.getMessage(), e); + + // ⚠️ 关键保护：无论什么异常，都确保消息不丢失 + boolean messageHandled = false; + + try { + // 检查是否为 InfluxDB 超时或连接异常 + boolean isInfluxTimeout = isInfluxDBTimeoutException(e); + + if (isInfluxTimeout) { + LOG.warn("检测到 InfluxDB 超时/连接异常，将消息发送到备份队列: {}", e.getMessage()); + amqpService.SendMessage("shengdilandevicedataback", message); + LOG.info("消息已发送到备份队列，确认原消息"); + messageHandled = true; + } else { + LOG.error("非超时异常，消息重新入队等待重试: {}", e.getMessage()); + channel.basicNack(deliveryTag, false, true); + messageHandled = true; + return; // 避免重复确认 + } + } catch (Exception sendError) { + // ⚠️⚠️⚠️ 最关键的保护：如果发送备份队列也失败，不要确认原消息 + LOG.error("发送备份队列失败，原消息将重新入队: {}", sendError.getMessage(), sendError); + channel.basicNack(deliveryTag, false, true); + messageHandled = true; + return; + } + + // 只有成功处理后，才确认原消息 + if (messageHandled) { + channel.basicAck(deliveryTag, false); + } } } +// @RabbitListener(queues = "shengdilandevicedataback") + public void deviceDataMqListenerForBackup(@Payload String message, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Channel channel) throws Exception { + if (StringUtils.isEmpty(message)) { + channel.basicAck(deliveryTag, false); + return; + } + + try { + History history = JSON.parseObject(message, History.class); + deviceDataService.saveDeviceDataToInfluxdb(history); + + // 成功写入后确认消息 + channel.basicAck(deliveryTag, false); + } catch (JSONException je) { + // ⚠️ 毒丸保护：JSON解析异常不会通过重试恢复，直接丢弃消息并记录日志 + LOG.error("JSON解析异常（不可恢复），消息将被丢弃: {}", je.getMessage(), je); + LOG.error("被丢弃的消息内容: {}", message); + channel.basicAck(deliveryTag, false); + } catch (Exception e) { + LOG.error("处理设备数据时发生异常: {}", e.getMessage(), e); + + // ⚠️ 关键保护：无论什么异常，都确保消息不丢失 + boolean messageHandled = false; + + try { + // 检查是否为 InfluxDB 超时或连接异常 + boolean isInfluxTimeout = isInfluxDBTimeoutException(e); + if (isInfluxTimeout) { + LOG.warn("检测到 InfluxDB 超时/连接异常，将消息发送到备份队列: {}", e.getMessage()); + amqpService.SendMessage("shengdilandevicedataback", message); + LOG.info("消息已发送到备份队列，确认原消息"); + messageHandled = true; + } else { + LOG.error("非超时异常，消息重新入队等待重试: {}", e.getMessage()); + channel.basicNack(deliveryTag, false, true); + messageHandled = true; + return; // 避免重复确认 + } + } catch (Exception sendError) { + // ⚠️⚠️⚠️ 最关键的保护：如果发送备份队列也失败，不要确认原消息 + LOG.error("发送备份队列失败，原消息将重新入队: {}", sendError.getMessage(), sendError); + channel.basicNack(deliveryTag, false, true); + messageHandled = true; + return; + } + + // 只有成功处理后，才确认原消息 + if (messageHandled) { + channel.basicAck(deliveryTag, false); + } + } + } /** * 监听设备下发命令日志 * @@ -62,23 +154,56 @@ public class AmqpListener { * @param channel * @throws Exception */ - - @RabbitListener(queues = "shengdilandevicelogall") +// @RabbitListener(queues = "shengdilandevicelogall") public void deviceLogMqListener(@Payload String message, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Channel channel) throws Exception { if (StringUtils.isEmpty(message)) { channel.basicAck(deliveryTag, false); return; } + try { DeviceLogData deviceLogData = JSON.parseObject(message, DeviceLogData.class); deviceLogService.saveDeviceLogToInfluxdb(deviceLogData); - LOG.info("saveDeviceLogToInfluxdb OK:" + deviceLogData.getImei()); + + // 成功写入后确认消息 + channel.basicAck(deliveryTag, false); + } catch (JSONException je) { + // ⚠️ 毒丸保护：JSON解析异常不会通过重试恢复，直接丢弃消息并记录日志 + LOG.error("JSON解析异常（不可恢复），消息将被丢弃: {}", je.getMessage(), je); + LOG.error("被丢弃的消息内容: {}", message); channel.basicAck(deliveryTag, false); } catch (Exception e) { - channel.basicNack(deliveryTag, false, true); - LOG.info("saveDeviceLogToInfluxdb Error:(" + e + ")-" + message); + LOG.error("处理设备日志时发生异常: {}", e.getMessage(), e); + + boolean messageHandled = false; + + try { + boolean isInfluxTimeout = isInfluxDBTimeoutException(e); + + if (isInfluxTimeout) { + LOG.warn("检测到 InfluxDB 超时/连接异常，将消息发送到备份队列: {}", e.getMessage()); + amqpService.SendMessage("shengdilandevicelogback", message); + LOG.info("消息已发送到备份队列，确认原消息"); + messageHandled = true; + } else { + LOG.error("非超时异常，消息重新入队等待重试: {}", e.getMessage()); + channel.basicNack(deliveryTag, false, true); + messageHandled = true; + return; + } + } catch (Exception sendError) { + LOG.error("发送备份队列失败，原消息将重新入队: {}", sendError.getMessage(), sendError); + channel.basicNack(deliveryTag, false, true); + messageHandled = true; + return; + } + + if (messageHandled) { + channel.basicAck(deliveryTag, false); + } } } + /** * 更新设备下发命令日志 * @@ -87,21 +212,109 @@ public class AmqpListener { * @param channel * @throws Exception */ - @RabbitListener(queues = "shengdilandevicelogupdate") public void updateDeviceLogMqListener(@Payload String message, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Channel channel) throws Exception { if (StringUtils.isEmpty(message)) { channel.basicAck(deliveryTag, false); return; } + try { DeviceLogData deviceLogData = JSON.parseObject(message, DeviceLogData.class); deviceLogService.updateDeviceLogMqListener(deviceLogData); - LOG.info("saveDeviceLogToInfluxdb OK:" + deviceLogData.getImei()); + + // 成功写入后确认消息 + channel.basicAck(deliveryTag, false); + } catch (JSONException je) { + // ⚠️ 毒丸保护：JSON解析异常不会通过重试恢复，直接丢弃消息并记录日志 + LOG.error("JSON解析异常（不可恢复），消息将被丢弃: {}", je.getMessage(), je); + LOG.error("被丢弃的消息内容: {}", message); channel.basicAck(deliveryTag, false); } catch (Exception e) { - channel.basicNack(deliveryTag, false, true); - LOG.info("saveDeviceLogToInfluxdb Error:(" + e + ")-" + message); + LOG.error("更新设备日志时发生异常: {}", e.getMessage(), e); + + boolean messageHandled = false; + + try { + boolean isInfluxTimeout = isInfluxDBTimeoutException(e); + + if (isInfluxTimeout) { + LOG.warn("检测到 InfluxDB 超时/连接异常，将消息发送到备份队列: {}", e.getMessage()); + amqpService.SendMessage("shengdilandevicelogback", message); + LOG.info("消息已发送到备份队列，确认原消息"); + messageHandled = true; + } else { + LOG.error("非超时异常，消息重新入队等待重试: {}", e.getMessage()); + channel.basicNack(deliveryTag, false, true); + messageHandled = true; + return; + } + } catch (Exception sendError) { + LOG.error("发送备份队列失败，原消息将重新入队: {}", sendError.getMessage(), sendError); + channel.basicNack(deliveryTag, false, true); + messageHandled = true; + return; + } + + if (messageHandled) { + channel.basicAck(deliveryTag, false); + } + } + } + + /** + * 检查异常是否为 InfluxDB 超时、连接失败或 shard 损坏异常 + * @param e 异常对象 + * @return 如果是超时、连接失败或 shard 损坏异常返回 true，否则返回 false + */ + private boolean isInfluxDBTimeoutException(Exception e) { + String message = e.getMessage(); + + // 检查 shard 损坏相关的错误 + if (message != null && (message.contains("not attempting to open shard") || + message.contains("short buffer") || + message.contains("InternalServerErrorException"))) { + return true; + } + + // 检查连接失败相关的错误 + if (message != null && (message.contains("Failed to connect") || + message.contains("ConnectException") || + message.contains("Connection refused") || + message.contains("Connection timed out"))) { + return true; + } + + // 检查直接异常类型 + if (e instanceof com.influxdb.exceptions.InfluxException) { + if (message != null && (message.contains("Read timed out") || + message.contains("SocketTimeoutException") || + message.contains("timeout") || + message.contains("connect timed out"))) { + return true; + } + } + + // 检查根本原因（使用HashSet防止循环引用导致死循环） + Throwable cause = e.getCause(); + Set visitedCauses = new HashSet<>(); + while (cause != null && visitedCauses.add(cause)) { + if (cause instanceof java.net.SocketTimeoutException || + cause instanceof java.net.ConnectException) { + return true; + } + if (cause.getMessage() != null && + (cause.getMessage().contains("Read timed out") || + cause.getMessage().contains("SocketTimeoutException") || + cause.getMessage().contains("connect timed out") || + cause.getMessage().contains("Failed to connect") || + cause.getMessage().contains("ConnectException") || + cause.getMessage().contains("Connection timed out"))) { + return true; + } + cause = cause.getCause(); } + + return false; } } diff --git a/src/main/java/com/topsail/influxdb/rabbitmq/service/AmqpService.java b/src/main/java/com/topsail/influxdb/rabbitmq/service/AmqpService.java index a076c0e..8553e35 100644 --- a/src/main/java/com/topsail/influxdb/rabbitmq/service/AmqpService.java +++ b/src/main/java/com/topsail/influxdb/rabbitmq/service/AmqpService.java @@ -1,46 +1,255 @@ package com.topsail.influxdb.rabbitmq.service; +import com.alibaba.fastjson.JSONArray; +import com.alibaba.fastjson.JSONObject; +import com.rabbitmq.client.Channel; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.amqp.core.AmqpAdmin; import org.springframework.amqp.core.AmqpTemplate; import org.springframework.amqp.core.Message; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.net.HttpURLConnection; +import java.net.URL; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Base64; +import java.util.List; +import java.util.concurrent.TimeoutException; + @Service public class AmqpService { public static final Logger LOG = LoggerFactory.getLogger(AmqpService.class); private final AmqpAdmin amqpAdmin; private final AmqpTemplate amqpTemplate; + private final ConnectionFactory connectionFactory; + + // 从配置文件读取 RabbitMQ Management API 配置 + @Value("${rabbitmq.management.url:http://localhost:15672}") + private String managementUrl; + + @Value("${rabbitmq.management.username:guest}") + private String managementUsername; + + @Value("${rabbitmq.management.password:guest}") + private String managementPassword; + + @Value("${rabbitmq.management.vhost:/}") + private String managementVhost; + + // 缓存认证头，避免重复计算（使用volatile保证可见性） + private volatile String cachedAuthHeader = null; @Autowired - public AmqpService(AmqpAdmin amqpAdmin, AmqpTemplate amqpTemplate) { + public AmqpService(AmqpAdmin amqpAdmin, AmqpTemplate amqpTemplate, ConnectionFactory connectionFactory) { this.amqpAdmin = amqpAdmin; this.amqpTemplate = amqpTemplate; + this.connectionFactory = connectionFactory; } - public void SendMessage(String queue,String content){ - //LOG.info("Send ampg" + content); - this.amqpTemplate.convertAndSend(queue,content); + public void SendMessage(String queue, String content) { + this.amqpTemplate.convertAndSend(queue, content); } - public void SendExchange(String exchange,String content){ - this.amqpTemplate.convertAndSend(exchange,"",content); + public void SendExchange(String exchange, String content) { + this.amqpTemplate.convertAndSend(exchange, "", content); } - public boolean IsQueuesEmpty(String queue){ - try{ + public boolean IsQueuesEmpty(String queue) { + try { Message msg = this.amqpTemplate.receive(queue); - if(msg!=null){ - this.amqpTemplate.send(queue,msg); + if (msg != null) { + this.amqpTemplate.send(queue, msg); return false; - }else{ + } else { return true; } - }catch (Exception e){ + } catch (Exception e) { return true; } } + + /** + * 获取指定队列的消息数量（兼容 Spring Boot 2.1.8） + * @param queueName 队列名称 + * @return 队列中待消费的消息数量，如果获取失败或队列不存在则返回-1 + */ + public long getQueueMessageCount(String queueName) { + if (queueName == null || queueName.trim().isEmpty()) { + LOG.warn("队列名称不能为空"); + return -1; + } + + org.springframework.amqp.rabbit.connection.Connection connection = null; + Channel channel = null; + + try { + connection = connectionFactory.createConnection(); + channel = connection.createChannel(false); + + // queueDeclarePassive 不会创建队列，只是获取队列信息，如果队列不存在会抛出异常 + com.rabbitmq.client.AMQP.Queue.DeclareOk declareOk = channel.queueDeclarePassive(queueName); + long messageCount = declareOk.getMessageCount(); + LOG.debug("队列 [{}] 当前消息数量: {}", queueName, messageCount); + return messageCount; + + } catch (IOException e) { + String errorMsg = e.getMessage(); + if (errorMsg != null && (errorMsg.contains("NOT_FOUND") || errorMsg.contains("404"))) { + LOG.warn("队列 [{}] 不存在", queueName); + } else { + LOG.error("获取队列 [{}] 消息数量时发生IO异常: {}", queueName, errorMsg); + } + return -1; + } catch (Exception e) { + LOG.error("获取队列 [{}] 消息数量时发生异常", queueName, e); + return -1; + } finally { + if (channel != null && channel.isOpen()) { + try { + channel.close(); + } catch (IOException | TimeoutException e) { + LOG.warn("关闭 Channel 时发生异常", e); + } + } + if (connection != null) { + try { + connection.close(); + } catch (Exception e) { + LOG.warn("关闭 Connection 时发生异常", e); + } + } + } + } + + /** + * 获取RabbitMQ中所有的队列名称（通过 Management HTTP API） + * @return 队列名称列表，如果获取失败则返回空列表 + */ + public List getAllQueueNames() { + List queueNames = new ArrayList<>(); + + String baseUrl = managementUrl.endsWith("/") ? managementUrl.substring(0, managementUrl.length() - 1) : managementUrl; + String apiUrl = baseUrl + "/api/queues"; + + if (managementVhost != null && !managementVhost.trim().isEmpty() && !"/".equals(managementVhost)) { + try { + String encodedVhost = java.net.URLEncoder.encode(managementVhost, StandardCharsets.UTF_8.name()); + apiUrl = baseUrl + "/api/queues/" + encodedVhost; + } catch (Exception e) { + LOG.warn("虚拟主机名称编码失败，使用默认队列列表", e); + } + } + + HttpURLConnection connection = null; + InputStream inputStream = null; + BufferedReader reader = null; + + try { + LOG.debug("正在连接 RabbitMQ Management API: {}", apiUrl); + + URL url = new URL(apiUrl); + connection = (HttpURLConnection) url.openConnection(); + + String authHeader = getAuthHeader(); + connection.setRequestProperty("Authorization", authHeader); + connection.setRequestProperty("Accept", "application/json"); + connection.setRequestMethod("GET"); + connection.setConnectTimeout(5000); + connection.setReadTimeout(10000); + + int responseCode = connection.getResponseCode(); + if (responseCode == HttpURLConnection.HTTP_OK) { + inputStream = connection.getInputStream(); + reader = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8)); + + StringBuilder response = new StringBuilder(); + char[] buffer = new char[4096]; + int bytesRead; + while ((bytesRead = reader.read(buffer)) != -1) { + response.append(buffer, 0, bytesRead); + } + + String jsonResponse = response.toString().trim(); + if (!jsonResponse.isEmpty()) { + try { + JSONArray queues = JSONArray.parseArray(jsonResponse); + if (queues != null && !queues.isEmpty()) { + for (int i = 0; i < queues.size(); i++) { + JSONObject queue = queues.getJSONObject(i); + String queueName = queue.getString("name"); + if (queueName != null && !queueName.trim().isEmpty()) { + queueNames.add(queueName); + } + } + LOG.info("成功获取 {} 个队列名称", queueNames.size()); + } else { + LOG.info("RabbitMQ 中暂无队列"); + } + } catch (Exception jsonEx) { + LOG.error("解析 RabbitMQ Management API 响应失败", jsonEx); + } + } else { + LOG.warn("RabbitMQ Management API 返回空响应"); + } + } else if (responseCode == HttpURLConnection.HTTP_UNAUTHORIZED) { + LOG.error("RabbitMQ Management API 认证失败，请检查用户名和密码配置"); + } else if (responseCode == HttpURLConnection.HTTP_NOT_FOUND) { + LOG.error("RabbitMQ Management API 未找到，请确认 rabbitmq_management 插件已启用"); + } else { + LOG.warn("RabbitMQ Management API 响应码: {}, 消息: {}", responseCode, connection.getResponseMessage()); + } + + } catch (IOException e) { + LOG.error("获取队列列表时发生IO异常，请确认 RabbitMQ Management 插件已启用且服务正常运行", e); + } catch (Exception e) { + LOG.error("获取队列列表时发生未知异常", e); + } finally { + if (reader != null) { + try { + reader.close(); + } catch (IOException e) { + LOG.warn("关闭 BufferedReader 时发生异常", e); + } + } + if (inputStream != null) { + try { + inputStream.close(); + } catch (IOException e) { + LOG.warn("关闭 InputStream 时发生异常", e); + } + } + if (connection != null) { + connection.disconnect(); + } + } + + return queueNames; + } + + /** + * 获取或生成 Basic Auth 认证头 + * @return Authorization header 值 + */ + private String getAuthHeader() { + if (cachedAuthHeader == null) { + synchronized (this) { + if (cachedAuthHeader == null) { + String auth = managementUsername + ":" + managementPassword; + String encodedAuth = Base64.getEncoder().encodeToString(auth.getBytes(StandardCharsets.UTF_8)); + cachedAuthHeader = "Basic " + encodedAuth; + } + } + } + return cachedAuthHeader; + } } diff --git a/src/main/java/com/topsail/influxdb/service/DeviceDataService.java b/src/main/java/com/topsail/influxdb/service/DeviceDataService.java index 012b7a1..8486ff2 100644 --- a/src/main/java/com/topsail/influxdb/service/DeviceDataService.java +++ b/src/main/java/com/topsail/influxdb/service/DeviceDataService.java @@ -5,10 +5,10 @@ import com.alibaba.fastjson.JSONObject; import com.influxdb.client.DeleteApi; import com.influxdb.client.InfluxDBClient; import com.influxdb.client.InfluxDBClientFactory; -import com.influxdb.client.WriteApi; import com.influxdb.client.domain.WritePrecision; import com.influxdb.query.FluxRecord; import com.influxdb.query.FluxTable; +import com.topsail.influxdb.config.InfluxDBConfig; import com.topsail.influxdb.entity.*; import com.topsail.influxdb.mapper.DeviceInfoMapper; import com.topsail.influxdb.pojo.History; @@ -33,7 +33,8 @@ import java.util.*; @Service public class DeviceDataService { public static final Logger LOG = LoggerFactory.getLogger(DeviceDataService.class); - public static final SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); + // 使用 ThreadLocal 解决 SimpleDateFormat 线程安全问题 + private static final ThreadLocal dateFormatLocal = ThreadLocal.withInitial(() -> new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")); // InfluxDB基础配置 private static final String DEVICEDATA_BUCKET_NAME = "iot"; @Value("${shengdilan.influxdb.token}") @@ -54,24 +55,54 @@ public class DeviceDataService { DeviceInfoMapper deviceInfoMapper; @Autowired AmqpService amqpService; - private InfluxDBClient influxDBClient; + @Autowired + InfluxDBClient influxDBClient; + @Autowired + InfluxDBConfig influxDBConfig; + + // 旧InfluxDB客户端（用于数据迁移） private InfluxDBClient oldInfluxDBClient; + /** 备份队列名称 */ + private static final String BACKUP_QUEUE_NAME = "shengdilandevicedataback"; + @PostConstruct public void init() { - this.influxDBClient = InfluxDBClientFactory.create(url, token.toCharArray(), org); this.oldInfluxDBClient = InfluxDBClientFactory.create(oldurl, oldtoken.toCharArray(), oldorg); + // 注册连接恢复回调：当InfluxDB恢复时，报告备份队列积压情况 + influxDBConfig.addRecoveryCallback(this::onConnectionRecovered); + LOG.info("DeviceDataService初始化完成，已注册InfluxDB连接恢复回调"); } @PreDestroy public void destroy() { - if (influxDBClient != null) { - influxDBClient.close(); - } if (oldInfluxDBClient != null) { - oldInfluxDBClient.close(); + try { + oldInfluxDBClient.close(); + } catch (Exception e) { + LOG.warn("关闭旧InfluxDB客户端时发生异常: {}", e.getMessage()); + } } } + + /** + * InfluxDB连接恢复后的回调处理 + * 报告备份队列积压情况，@RabbitListener会自动重试处理备份队列中的消息 + */ + private void onConnectionRecovered() { + try { + long backupQueueSize = amqpService.getQueueMessageCount(BACKUP_QUEUE_NAME); + if (backupQueueSize > 0) { + LOG.info("InfluxDB连接已恢复，备份队列 [{}] 当前积压{}条消息，@RabbitListener将自动重试处理", + BACKUP_QUEUE_NAME, backupQueueSize); + } else { + LOG.info("InfluxDB连接已恢复，备份队列无积压消息"); + } + } catch (Exception e) { + LOG.warn("InfluxDB连接已恢复，但检查备份队列积压情况时发生异常: {}", e.getMessage()); + } + } + /** * 根据设备号查询设备历史数据 * @@ -85,60 +116,61 @@ public class DeviceDataService { */ public List getDeviceHistoryData(String uid, Integer pageNo, Integer pageSize, String startTime, String endTime, String imei) { InfluxDBClient client = InfluxDBClientFactory.create(url, token.toCharArray(), org); - StringBuffer query = new StringBuffer(); - query.append("from(bucket: \"iot\") "); - if (startTime != null && endTime != null) { - SimpleDateFormat oldFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); - SimpleDateFormat newFormat = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'"); - newFormat.setTimeZone(TimeZone.getTimeZone("UTC"));//时区转换 - String start = null; - try { - start = newFormat.format(oldFormat.parse(startTime)); - String stop = newFormat.format(oldFormat.parse(endTime)); - query.append(String.format(" |> range(start:%s, stop:%s)", start, stop)); - } catch (ParseException e) { - e.printStackTrace(); + try { + StringBuffer query = new StringBuffer(); + query.append("from(bucket: \"iot\") "); + if (startTime != null && endTime != null) { + SimpleDateFormat oldFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); + SimpleDateFormat newFormat = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'"); + newFormat.setTimeZone(TimeZone.getTimeZone("UTC"));//时区转换 + String start = null; + try { + start = newFormat.format(oldFormat.parse(startTime)); + String stop = newFormat.format(oldFormat.parse(endTime)); + query.append(String.format(" |> range(start:%s, stop:%s)", start, stop)); + } catch (ParseException e) { + e.printStackTrace(); + } + } else { + query.append("|> range(start: -1d)"); } - } else { - query.append("|> range(start: -1d)"); - } - query.append(String.format(" |> filter(fn: (r) => r[\"_measurement\"] == \"history\") |> filter(fn: (r) => r[\"imei\"] == \"%s\") |> filter(fn: (r) => r[\"_field\"] == \"jsondata\" ) |> sort(columns:[\"_time\"], desc:true) ", imei)); - if (pageNo != null && pageSize != null) { - query.append(String.format(" |> limit(n: %s,offset:%s)", pageSize, pageNo - 1)); - } - query.append(" |> yield(name: \"last\")"); - System.out.println("查询语句==========:" + query); - List tables = client.getQueryApi().query(query.toString()); -// List returnList = new ArrayList<>(); - List resultSet = new ArrayList<>(); - for (FluxTable fluxTable : tables) { - List records = fluxTable.getRecords(); - for (FluxRecord fluxRecord : records) { - String value = (String) fluxRecord.getValueByKey("_value"); - DeviceHistoryData data = JSONObject.parseObject(value, DeviceHistoryData.class); - Integer singalstrength = (Integer) JSONObject.parseObject(value).get("singalstrength"); - data.setSignalStrength(singalstrength); - data.setTime(dateFormat.format((Long.parseLong(data.getTime())))); - DeviceHistoryVo dataVo = DeviceHistoryVo.builder() - .deviceType(String.valueOf(data.getDeviceType())) - .imei(data.getImei()) - .batteryLevel(data.getBatteryLevel()) - .singalStrength(data.getSignalStrength()) - .sampleData(data.getSampleData()) - .passNum(data.getPassNum()) - .alarmType(data.getAlarmType()) - .unit(data.getUnit()) - .sendTime(data.getTime()) - .dataBody(data.getDataBody()) - .value(data.getValue()).build(); - dataVo = analysisSampleData(dataVo, data.getValue(), data.getUnit()); - resultSet.add(dataVo); -// returnList.add(data); + query.append(String.format(" |> filter(fn: (r) => r[\"_measurement\"] == \"history\") |> filter(fn: (r) => r[\"imei\"] == \"%s\") |> filter(fn: (r) => r[\"_field\"] == \"jsondata\" ) |> sort(columns:[\"_time\"], desc:true) ", imei)); + if (pageNo != null && pageSize != null) { + query.append(String.format(" |> limit(n: %s,offset:%s)", pageSize, pageNo - 1)); + } + query.append(" |> yield(name: \"last\")"); + System.out.println("查询语句==========:" + query); + List tables = client.getQueryApi().query(query.toString()); + List resultSet = new ArrayList<>(); + for (FluxTable fluxTable : tables) { + List records = fluxTable.getRecords(); + for (FluxRecord fluxRecord : records) { + String value = (String) fluxRecord.getValueByKey("_value"); + DeviceHistoryData data = JSONObject.parseObject(value, DeviceHistoryData.class); + Integer singalstrength = (Integer) JSONObject.parseObject(value).get("singalstrength"); + data.setSignalStrength(singalstrength); + data.setTime(dateFormatLocal.get().format((Long.parseLong(data.getTime())))); + DeviceHistoryVo dataVo = DeviceHistoryVo.builder() + .deviceType(String.valueOf(data.getDeviceType())) + .imei(data.getImei()) + .batteryLevel(data.getBatteryLevel()) + .singalStrength(data.getSignalStrength()) + .sampleData(data.getSampleData()) + .passNum(data.getPassNum()) + .alarmType(data.getAlarmType()) + .unit(data.getUnit()) + .sendTime(data.getTime()) + .dataBody(data.getDataBody()) + .value(data.getValue()).build(); + dataVo = analysisSampleData(dataVo, data.getValue(), data.getUnit()); + resultSet.add(dataVo); + } } + Collections.reverse(resultSet); + return resultSet; + } finally { + client.close(); } - client.close(); - Collections.reverse(resultSet); - return resultSet; } /** @@ -184,17 +216,13 @@ public class DeviceDataService { */ public List getOldInfluxdbData(String imei) { DeviceBelongInfo deviceBelongInfo = deviceInfoMapper.queryDeviceBelongInfo(imei); -// InfluxDBClient client = InfluxDBClientFactory.create(oldurl, oldtoken.toCharArray(), oldorg); StringBuffer query = new StringBuffer(); query.append("from(bucket: \"iot\") "); query.append("|> range(start: -1y)"); query.append(String.format(" |> filter(fn: (r) => r[\"_measurement\"] == \"history\") |> filter(fn: (r) => r[\"imei\"] == \"%s\") |> filter(fn: (r) => r[\"_field\"] == \"jsondata\" ) |> sort(columns:[\"_time\"], desc:true) ", imei)); -// if (pageNo != null && pageSize != null) { -// query.append(String.format(" |> limit(n: %s,offset:%s)", pageSize, pageNo - 1)); -// } query.append(" |> yield(name: \"last\")"); System.out.println("查询语句==========:" + query); - if(oldInfluxDBClient==null){ + if (oldInfluxDBClient == null) { oldInfluxDBClient = InfluxDBClientFactory.create(oldurl, oldtoken.toCharArray(), oldorg); } List tables = oldInfluxDBClient.getQueryApi().query(query.toString()); @@ -204,10 +232,25 @@ public class DeviceDataService { List records = fluxTable.getRecords(); for (FluxRecord fluxRecord : records) { String value = (String) fluxRecord.getValueByKey("_value"); - History history = JSONObject.parseObject(value, History.class); + History history = null; + try { + history = JSONObject.parseObject(value, History.class); + } catch (Exception e) { + LOG.warn("直接解析历史数据JSON失败，尝试清理后解析。IMEI: {}, 错误: {}", imei, e.getMessage()); + try { + String cleanedValue = clearDataBodyValue(value); + history = JSONObject.parseObject(cleanedValue, History.class); + String databody = extractAndClearDataBody(value); + history.setDatabody(databody); + LOG.info("成功清理并解析历史数据JSON。IMEI: {}", imei); + } catch (Exception cleanEx) { + LOG.error("清理后仍无法解析历史数据JSON，跳过该条记录。IMEI: {}, 原始数据前200字符: {}", + imei, value != null && value.length() > 200 ? value.substring(0, 200) + "..." : value); + continue; + } + } if (history != null) { Date createtime = history.getSenddate(); - //将时间转换成Instant Instant time = null; if (createtime != null) { time = createtime.toInstant(); @@ -221,7 +264,6 @@ public class DeviceDataService { } } } -// client.close(); Collections.reverse(rerurnList); return rerurnList; } @@ -232,14 +274,21 @@ public class DeviceDataService { * @param history */ public void saveDeviceDataToInfluxdb(History history) { + // 先进行null检查，避免空指针异常 + if (history == null || (StringUtils.isEmpty(history.getImei()))) { + return; + } if (history.getSenddate() == null) { return; } - if (history == null || (StringUtils.isEmpty(history.getImei()))) { + // 检查InfluxDB连接是否可用，不可用时直接将数据发送到MQ备份 + if (!influxDBConfig.isConnectionHealthy()) { + LOG.warn("InfluxDB连接不可用，数据直接发送到MQ备份。设备: {}", history.getImei()); + amqpService.SendMessage("shengdilandevicedataback", JSON.toJSONString(history)); return; } //查询设备的所属信息 - if (history != null && (history.getDeviceBelongInfo() == null || history.getHouseId() == null)) { + if (history.getDeviceBelongInfo() == null || history.getHouseId() == null) { DeviceBelongInfo deviceBelongInfo = getDeviceBelongInfo(history.getImei().trim()); if (deviceBelongInfo == null) { return; @@ -248,21 +297,72 @@ public class DeviceDataService { history.setHouseId(deviceBelongInfo.getHouseId()); } Date createtime = history.getSenddate(); - //将时间转换成Instant Instant time = null; if (createtime != null) { time = createtime.toInstant(); } - // 复用已创建的客户端实例 DeviceDataInfluxData deviceDataInfluxData = rebuildDeviceDataInfluxData(history, time); - try (WriteApi writeApi = influxDBClient.getWriteApi()) { - writeApi.writeMeasurement(DEVICEDATA_BUCKET_NAME, org, WritePrecision.NS, deviceDataInfluxData); - } catch (Exception e) { - e.printStackTrace(); - LOG.error("设备数据写入influxdb失败：{}", history.getImei()); - amqpService.SendMessage("shengdilandevicedataback", JSON.toJSONString(history)); + + int maxRetries = 3; + boolean success = false; + for (int retry = 0; retry < maxRetries && !success; retry++) { + try { + influxDBConfig.getCurrentWriteApi().writeMeasurement(DEVICEDATA_BUCKET_NAME, org, WritePrecision.NS, deviceDataInfluxData); + success = true; + LOG.info("设备数据写入influxdb成功：{}", history.getImei()); + } catch (Exception e) { + LOG.error("设备数据写入influxdb失败（重试 {}/{}）：{}，错误：{}", + retry + 1, maxRetries, history.getImei(), e.getMessage()); + + boolean isTimeout = isInfluxDBTimeoutException(e); + + // 如果是服务器内部错误（如shard损坏），不重试，直接发送到MQ + if (e.getMessage() != null && (e.getMessage().contains("InternalServerErrorException") || + e.getMessage().contains("not attempting to open shard") || + e.getMessage().contains("short buffer"))) { + LOG.error("InfluxDB服务器内部错误，可能是shard损坏，数据发送到MQ备份。错误详情: {}", e.getMessage()); + // 标记连接不健康，避免其他线程继续尝试无效写入 + influxDBConfig.markConnectionUnhealthy(); + try { + amqpService.SendMessage("shengdilandevicedataback", JSON.toJSONString(history)); + } catch (Exception sendEx) { + LOG.error("发送备份队列失败，异常将抛回给MQ监听器，消息将重新入队: {}", sendEx.getMessage()); + throw new RuntimeException("发送备份队列失败", sendEx); + } + return; + } + + // 如果是超时异常，立即标记连接不健康，发送到备份队列 + if (isTimeout) { + LOG.warn("检测到 InfluxDB 超时异常，标记连接不健康，将数据发送到备份队列: {}", e.getMessage()); + influxDBConfig.markConnectionUnhealthy(); + try { + amqpService.SendMessage("shengdilandevicedataback", JSON.toJSONString(history)); + } catch (Exception sendEx) { + LOG.error("发送备份队列失败，异常将抛回给MQ监听器，消息将重新入队: {}", sendEx.getMessage()); + throw new RuntimeException("发送备份队列失败", sendEx); + } + return; + } + + // 最后一次重试失败，发送到MQ + if (retry == maxRetries - 1) { + LOG.error("设备数据写入influxdb最终失败：{}", history.getImei()); + try { + amqpService.SendMessage("shengdilandevicedataback", JSON.toJSONString(history)); + } catch (Exception sendEx) { + LOG.error("发送备份队列失败，异常将抛回给MQ监听器，消息将重新入队: {}", sendEx.getMessage()); + throw new RuntimeException("发送备份队列失败", sendEx); + } + } else { + try { + Thread.sleep(2000 * (retry + 1)); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + } + } + } } - LOG.info("设备数据写入influxdb成功：{}", history.getImei()); } /** @@ -281,23 +381,22 @@ public class DeviceDataService { deviceDataInfluxData.value3 = 0; deviceDataInfluxData.value4 = 0; try { - if (values.length > 0) { - deviceDataInfluxData.value1 = Double.parseDouble(values[0]); + if (values.length > 0 && values[0] != null && !"null".equals(values[0])) { + deviceDataInfluxData.value1 = parseDoubleValue(values[0]); } - if (values.length > 1) { - deviceDataInfluxData.value2 = Double.parseDouble(values[1]); + if (values.length > 1 && values[1] != null && !"null".equals(values[1])) { + deviceDataInfluxData.value2 = parseDoubleValue(values[1]); } - if (values.length > 2) { - deviceDataInfluxData.value3 = Double.parseDouble(values[2]); + if (values.length > 2 && values[2] != null && !"null".equals(values[2])) { + deviceDataInfluxData.value3 = parseDoubleValue(values[2]); } - if (values.length > 3) { - deviceDataInfluxData.value4 = Double.parseDouble(values[4]); + if (values.length > 3 && values[3] != null && !"null".equals(values[3])) { + deviceDataInfluxData.value4 = parseDoubleValue(values[3]); } } catch (Exception e) { - LOG.info(e.getMessage()); + LOG.warn("解析设备数据值失败，imei: {}, value: {}, 错误: {}", history.getImei(), value, e.getMessage()); } } - //从内部移动处出来，value有没有数值都进行插入 deviceDataInfluxData.battery = history.getBatterylevel(); deviceDataInfluxData.sigal = history.getSingalstrength(); deviceDataInfluxData.jsondata = JSON.toJSONString(history); @@ -308,6 +407,32 @@ public class DeviceDataService { return deviceDataInfluxData; } + /** + * 解析数值字符串 + * 支持格式："0.2496Mpa", "25℃", "80%RH", "118.0", "-50.0", "null"等 + * - 包含单位(Mpa、℃、%RH)的字符串：提取数值部分 + * - 纯数字字符串：直接解析 + * + * @param value 待解析的字符串 + * @return 解析后的double值，解析失败返回0 + */ + private double parseDoubleValue(String value) { + if (value == null || value.trim().isEmpty() || "null".equals(value)) { + return 0; + } + try { + // 去除所有非数字、小数点、负号的字符（如单位：Mpa, ℃, %RH等） + String numericStr = value.replaceAll("[^\\d.\\-]", "").trim(); + if (numericStr.isEmpty()) { + return 0; + } + return Double.parseDouble(numericStr); + } catch (NumberFormatException e) { + LOG.debug("无法解析数值: {}", value); + return 0; + } + } + /** * 查询设备的所属信息 */ @@ -320,16 +445,10 @@ public class DeviceDataService { */ public void deleteDeviceData() { InfluxDBClient client = InfluxDBClientFactory.create(url, token.toCharArray()); - StringBuffer query = new StringBuffer(); - query.append("from(bucket: \"iot\") "); - query.append(String.format(" |> filter(fn: (r) => r[\"_measurement\"] == \"history\")")); - query.append("|> range(start: -36d)"); - System.out.println("查询语句==========:" + query); - DeleteApi deleteApi = client.getDeleteApi(); - OffsetDateTime start = OffsetDateTime.of(2020, 1, 1, 0, 0, 0, 0, ZoneOffset.UTC); - OffsetDateTime stop = OffsetDateTime.of(2026, 1, 1, 0, 0, 0, 0, ZoneOffset.UTC); try { - // Delete data with specific time range + DeleteApi deleteApi = client.getDeleteApi(); + OffsetDateTime start = OffsetDateTime.of(2020, 1, 1, 0, 0, 0, 0, ZoneOffset.UTC); + OffsetDateTime stop = OffsetDateTime.of(2026, 1, 1, 0, 0, 0, 0, ZoneOffset.UTC); deleteApi.delete(start, stop, "", DEVICEDATA_BUCKET_NAME, org); System.out.println("Data deleted successfully"); } catch (Exception e) { @@ -345,45 +464,69 @@ public class DeviceDataService { * @param imei */ public void transferDeviceData(String imei) { - //1.查询所有的设备编号 List syncDataFlags = deviceInfoMapper.querySyncDeviceDataFlagInfo(imei); - //2.根据设备编号查询历史Influxdb所有的设备数据 Boolean flag = true; if (syncDataFlags != null && syncDataFlags.size() > 0) { for (int i = 0; i < syncDataFlags.size(); i++) { if (!flag) { syncDataFlags = deviceInfoMapper.querySyncDeviceDataFlagInfo(null); + if (syncDataFlags == null || syncDataFlags.isEmpty()) { + LOG.warn("重新查询同步标志列表为空，终止同步"); + break; + } } int index = new Random().nextInt(syncDataFlags.size()); SyncDataFlag syncDataFlag = syncDataFlags.get(index); Boolean syncDeviceData = false; + Long dataCount = 0L; List influxdbDataList = getOldInfluxdbData(syncDataFlag.getImei().toLowerCase(Locale.ROOT)); if (influxdbDataList != null && influxdbDataList.size() > 0) { for (DeviceDataInfluxData influxData : influxdbDataList) { -// InfluxDBClient client = InfluxDBClientFactory.create(url, token.toCharArray()); - //3.将设备数据保存到influxdb中 - if(influxDBClient== null){ - influxDBClient = InfluxDBClientFactory.create(url, token.toCharArray()); - } - try (WriteApi writeApi = influxDBClient.getWriteApi()) { - writeApi.writeMeasurement(DEVICEDATA_BUCKET_NAME, org, WritePrecision.NS, influxData); - syncDeviceData = true; - } catch (Exception e) { - LOG.error("保存设备数据到influxdb失败：{}", e.getMessage()); - syncDeviceData = false; - if (influxData != null && influxData.jsondata != null && !influxData.jsondata.equals("")) { - String message = influxData.jsondata; - History history = JSON.parseObject(message, History.class); - if (history != null && history.getImei() != null && !history.getImei().equals("")) { - amqpService.SendMessage("shengdilandevicedataback", JSON.toJSONString(history)); + boolean writeSuccess = false; + int maxRetries = 3; + for (int retry = 0; retry < maxRetries && !writeSuccess; retry++) { + try { + influxDBConfig.getCurrentWriteApi().writeMeasurement(DEVICEDATA_BUCKET_NAME, org, WritePrecision.NS, influxData); + syncDeviceData = true; + dataCount = dataCount + 1; + writeSuccess = true; + LOG.info("设备数据保存到influxdb成功：{}", influxData.imei); + } catch (Exception e) { + LOG.error("保存设备数据到influxdb失败（重试 {}/{}）：{}，错误：{}", + retry + 1, maxRetries, influxData.imei, e.getMessage()); + + if (e.getMessage() != null && (e.getMessage().contains("InternalServerErrorException") || + e.getMessage().contains("not attempting to open shard") || + e.getMessage().contains("short buffer"))) { + LOG.error("InfluxDB服务器内部错误，可能是shard损坏，数据发送到MQ备份。错误详情: {}", e.getMessage()); + if (influxData != null && influxData.jsondata != null && !influxData.jsondata.equals("")) { + History history = JSON.parseObject(influxData.jsondata, History.class); + if (history != null && history.getImei() != null && !history.getImei().equals("")) { + amqpService.SendMessage("shengdilandevicedataback", JSON.toJSONString(history)); + } + } + break; } - } + if (retry == maxRetries - 1) { + if (influxData != null && influxData.jsondata != null && !influxData.jsondata.equals("")) { + History history = JSON.parseObject(influxData.jsondata, History.class); + if (history != null && history.getImei() != null && !history.getImei().equals("")) { + amqpService.SendMessage("shengdilandevicedataback", JSON.toJSONString(history)); + } + } + } else { + try { + Thread.sleep(2000 * (retry + 1)); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + } + } + } } } } if (syncDeviceData) { - //4.更新同步设备数据状态 deviceInfoMapper.updateSyncDeviceDataFlagInfo(syncDataFlag.getId(), 1); } flag = false; @@ -396,35 +539,126 @@ public class DeviceDataService { */ public Long getDeviceDataCount(String imei) { InfluxDBClient client = InfluxDBClientFactory.create(oldurl, oldtoken.toCharArray(), oldorg); - // 1. 拼接公共过滤条件片段（总条数和分页查询共用） - StringBuilder filterFragment = new StringBuilder(); - filterFragment.append("from(bucket: \"iot\") "); - filterFragment.append("|> range(start: -400d)"); - // 2. 拼接固定过滤条件（_measurement和_field，必选） - filterFragment.append(" |> filter(fn: (r) => r[\"_measurement\"] == \"history\")"); - filterFragment.append(" |> filter(fn: (r) => r[\"_field\"] == \"jsondata\")"); - // 3. 动态拼接imei过滤条件（仅当imei非空且非空白字符串时拼接） - if (imei != null && !imei.trim().isEmpty()) { - filterFragment.append(String.format(" |> filter(fn: (r) => r[\"imei\"] == \"%s\")", imei.trim())); - } - // 5. 拼接固定排序条件（按时间倒序） - // 2. 构建总条数查询子句（yield命名为total） - StringBuilder totalQuery = new StringBuilder(); - totalQuery.append(filterFragment); // 复用过滤条件 - totalQuery.append(" |> count(column: \"_value\")"); // 统计总条数 - totalQuery.append(" |> yield(name: \"total\")\n"); // 标记结果集为total - System.out.println("查询数量语句：" + totalQuery.toString()); - //查询总条数 - long totalCount = 0; - List totalTables = client.getQueryApi().query(totalQuery.toString()); - for (FluxTable table : totalTables) { - for (FluxRecord record : table.getRecords()) { - Object total = record.getValue(); - if (total != null) { - totalCount = totalCount + ((Number) total).longValue(); + try { + StringBuilder filterFragment = new StringBuilder(); + filterFragment.append("from(bucket: \"iot\") "); + filterFragment.append("|> range(start: -400d)"); + filterFragment.append(" |> filter(fn: (r) => r[\"_measurement\"] == \"history\")"); + filterFragment.append(" |> filter(fn: (r) => r[\"_field\"] == \"jsondata\")"); + if (imei != null && !imei.trim().isEmpty()) { + filterFragment.append(String.format(" |> filter(fn: (r) => r[\"imei\"] == \"%s\")", imei.trim())); + } + StringBuilder totalQuery = new StringBuilder(); + totalQuery.append(filterFragment); + totalQuery.append(" |> count(column: \"_value\")"); + totalQuery.append(" |> yield(name: \"total\")\n"); + System.out.println("查询数量语句：" + totalQuery.toString()); + long totalCount = 0; + List totalTables = client.getQueryApi().query(totalQuery.toString()); + for (FluxTable table : totalTables) { + for (FluxRecord record : table.getRecords()) { + Object total = record.getValue(); + if (total != null) { + totalCount = totalCount + ((Number) total).longValue(); + } } } + return totalCount; + } finally { + client.close(); + } + } + + /** + * 检查异常是否为 InfluxDB 超时、连接失败或 shard 损坏异常 + * @param e 异常对象 + * @return 如果是超时、连接失败或 shard 损坏异常返回 true，否则返回 false + */ + private boolean isInfluxDBTimeoutException(Exception e) { + String message = e.getMessage(); + + // 检查 shard 损坏相关的错误 + if (message != null && (message.contains("not attempting to open shard") || + message.contains("short buffer") || + message.contains("InternalServerErrorException"))) { + return true; + } + + // 检查连接失败相关的错误 + if (message != null && (message.contains("Failed to connect") || + message.contains("ConnectException") || + message.contains("Connection refused") || + message.contains("Connection timed out"))) { + return true; + } + + // 检查直接异常类型 + if (e instanceof com.influxdb.exceptions.InfluxException) { + if (message != null && (message.contains("Read timed out") || + message.contains("SocketTimeoutException") || + message.contains("timeout") || + message.contains("connect timed out"))) { + return true; + } + } + + // 检查根本原因（使用HashSet防止循环引用导致死循环） + Throwable cause = e.getCause(); + Set visitedCauses = new HashSet<>(); + while (cause != null && visitedCauses.add(cause)) { + if (cause instanceof java.net.SocketTimeoutException || + cause instanceof java.net.ConnectException) { + return true; + } + if (cause.getMessage() != null && + (cause.getMessage().contains("Read timed out") || + cause.getMessage().contains("SocketTimeoutException") || + cause.getMessage().contains("connect timed out") || + cause.getMessage().contains("Failed to connect") || + cause.getMessage().contains("ConnectException") || + cause.getMessage().contains("Connection timed out"))) { + return true; + } + cause = cause.getCause(); + } + + return false; + } + + /** + * 清理databody字段值（置为null）以尝试修复JSON解析问题 + */ + private String clearDataBodyValue(String value) { + if (value == null || value.isEmpty()) { + return value; + } + try { + // 使用正则替换databody字段值为null + return value.replaceAll("\"databody\"\\s*:\\s*\"[^\"]*\"", "\"databody\":null"); + } catch (Exception e) { + LOG.warn("清理databody字段失败: {}", e.getMessage()); + return value; + } + } + + /** + * 提取并清理databody字段内容 + * 使用正则从原始字符串中提取databody值，避免JSON解析失败的问题 + */ + private String extractAndClearDataBody(String value) { + if (value == null || value.isEmpty()) { + return null; + } + try { + // 使用正则从原始字符串中提取databody字段值 + java.util.regex.Pattern pattern = java.util.regex.Pattern.compile("\"databody\"\\s*:\\s*\"((?:[^\"\\\\]|\\\\.)*)\""); + java.util.regex.Matcher matcher = pattern.matcher(value); + if (matcher.find()) { + return matcher.group(1); + } + } catch (Exception e) { + LOG.warn("提取databody字段失败: {}", e.getMessage()); } - return totalCount; + return null; } } diff --git a/src/main/java/com/topsail/influxdb/service/DeviceLogService.java b/src/main/java/com/topsail/influxdb/service/DeviceLogService.java index 6cc4b9e..564f7d1 100644 --- a/src/main/java/com/topsail/influxdb/service/DeviceLogService.java +++ b/src/main/java/com/topsail/influxdb/service/DeviceLogService.java @@ -5,10 +5,10 @@ import com.alibaba.fastjson.JSONObject; import com.influxdb.client.DeleteApi; import com.influxdb.client.InfluxDBClient; import com.influxdb.client.InfluxDBClientFactory; -import com.influxdb.client.WriteApi; import com.influxdb.client.domain.WritePrecision; import com.influxdb.query.FluxRecord; import com.influxdb.query.FluxTable; +import com.topsail.influxdb.config.InfluxDBConfig; import com.topsail.influxdb.entity.*; import com.topsail.influxdb.mapper.DeviceInfoMapper; import com.topsail.influxdb.rabbitmq.service.AmqpService; @@ -34,7 +34,7 @@ public class DeviceLogService { public static final Logger LOG = LoggerFactory.getLogger(DeviceLogService.class); private static volatile Map supplierInfoMap = new HashMap<>(); - // 初始化时加载圣地蓝项目信息（例如，在构造函数或@PostConstruct方法中） + // 初始化时加载圣地蓝项目信息 @PostConstruct public void initShendianlanProjects() { if (supplierInfoMap.isEmpty()) { @@ -42,7 +42,6 @@ public class DeviceLogService { if (supplierInfoMap.isEmpty()) { List supplierList = deviceInfoMapper.getSupplierList(); if (supplierList != null && !supplierList.isEmpty()) { - //查询出属于圣地蓝项目的IMEI号 for (SupplierVO supplierVO : supplierList) { Integer id = supplierVO.getId(); String name = supplierVO.getSupplierName(); @@ -70,6 +69,8 @@ public class DeviceLogService { DeviceInfoMapper deviceInfoMapper; @Autowired AmqpService amqpService; + @Autowired + InfluxDBConfig influxDBConfig; /** * 存储设备数据到influxdb @@ -84,13 +85,18 @@ public class DeviceLogService { if (createtime == null) { createtime = new Date(); } - //将时间转换成Instant Instant time = createtime.toInstant(); - InfluxDBClient client = InfluxDBClientFactory.create(url, token.toCharArray()); + + // 检查InfluxDB连接是否可用，不可用时直接将数据发送到MQ备份 + if (!influxDBConfig.isConnectionHealthy()) { + LOG.warn("InfluxDB连接不可用，日志数据直接发送到MQ备份。设备: {}", deviceLogData.getImei()); + amqpService.SendMessage("shengdilandevicelogback", JSON.toJSONString(deviceLogData)); + return; + } + DeviceLogInfluxData deviceLogInfluxData = new DeviceLogInfluxData(); deviceLogInfluxData.id = deviceLogData.getId(); deviceLogInfluxData.imei = deviceLogData.getImei(); - //从内部移动处出来，value有没有数值都进行插入 deviceLogInfluxData.result = deviceLogData.getResult(); deviceLogInfluxData.statusIssue = deviceLogData.getStatusIssue(); deviceLogInfluxData.supplierId = deviceLogData.getSupplierId(); @@ -103,13 +109,47 @@ public class DeviceLogService { deviceLogInfluxData.deviceBelongInfo = deviceLogData.getDeviceBelongInfo(); deviceLogInfluxData.houseId = deviceLogData.getHouseId(); deviceLogInfluxData.time = time != null ? time : Instant.now(); - try (WriteApi writeApi = client.getWriteApi()) { - writeApi.writeMeasurement(LOG_BUCKET_NAME, org, WritePrecision.NS, deviceLogInfluxData); - } catch (Exception e) { - LOG.error("下发日志写入influxdb失败：{}", e.getMessage()); - amqpService.SendMessage("shengdilandevicelogback", JSON.toJSONString(deviceLogData)); + + int maxRetries = 3; + boolean success = false; + for (int retry = 0; retry < maxRetries && !success; retry++) { + try { + influxDBConfig.getCurrentWriteApi().writeMeasurement(LOG_BUCKET_NAME, org, WritePrecision.NS, deviceLogInfluxData); + success = true; + LOG.info("下发日志写入influxdb成功：{}", deviceLogData.getImei()); + } catch (Exception e) { + LOG.error("下发日志写入influxdb失败（重试 {}/{}）：{}，错误：{}", + retry + 1, maxRetries, deviceLogData.getImei(), e.getMessage()); + + // 如果是服务器内部错误（如shard损坏），不重试，直接发送到MQ + if (e.getMessage() != null && (e.getMessage().contains("InternalServerErrorException") || + e.getMessage().contains("not attempting to open shard") || + e.getMessage().contains("short buffer"))) { + LOG.error("InfluxDB服务器内部错误，可能是shard损坏，数据发送到MQ备份。错误详情: {}", e.getMessage()); + influxDBConfig.markConnectionUnhealthy(); + amqpService.SendMessage("shengdilandevicelogback", JSON.toJSONString(deviceLogData)); + return; + } + + // 如果是超时/连接异常，立即标记连接不健康 + if (isInfluxDBTimeoutException(e)) { + LOG.warn("检测到 InfluxDB 超时异常，标记连接不健康: {}", e.getMessage()); + influxDBConfig.markConnectionUnhealthy(); + } + + // 最后一次重试失败，发送到MQ + if (retry == maxRetries - 1) { + LOG.error("下发日志写入influxdb最终失败：{}", deviceLogData.getImei()); + amqpService.SendMessage("shengdilandevicelogback", JSON.toJSONString(deviceLogData)); + } else { + try { + Thread.sleep(2000 * (retry + 1)); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + } + } + } } - LOG.info("下发日志写入influxdb成功：{}", deviceLogData.getImei()); } /** @@ -119,9 +159,7 @@ public class DeviceLogService { Calendar calendar = Calendar.getInstance(); calendar.setTime(new Date()); InfluxDBClient client = InfluxDBClientFactory.create(url, token.toCharArray(), org); - // 1. 拼接公共过滤条件片段（总条数和分页查询共用） StringBuilder predicates = new StringBuilder(); - // 2. 拼接固定过滤条件（_measurement和_field，必选） predicates.append("_measurement=devicelog"); if (deviceLogId != null) { if (predicates.length() > 0) { @@ -131,7 +169,6 @@ public class DeviceLogService { System.out.println(predicates.toString()); DeleteApi deleteApi = client.getDeleteApi(); try { -// // 删除指定时间范围内所有数据 OffsetDateTime start = OffsetDateTime.of(calendar.get(Calendar.YEAR) - 10, 1, 1, 0, 0, 0, 0, ZoneOffset.UTC); OffsetDateTime stop = OffsetDateTime.of(calendar.get(Calendar.YEAR) + 1, 1, 1, 0, 0, 0, 0, ZoneOffset.UTC); deleteApi.delete(start, stop, predicates.toString(), LOG_BUCKET_NAME, org); @@ -143,9 +180,6 @@ public class DeviceLogService { } } - -// from(bucket: "devicelog") |> range(start: -3600d) |> filter(fn: (r) => r["_measurement"] == "devicelog") |> filter(fn: (r) => r["_field"] == "logjson") |> filter(fn: (r) => r["deviceBelongInfo"] =~ /.*白志.*/) |> sort(columns:["_time"], desc:true) |> limit(n: 10, offset: 0) |> yield(name: "data") - /** * 查询设备下发命令日志信息 * @@ -164,10 +198,11 @@ public class DeviceLogService { public JSONObject getPageDeviceLog(Integer pageNode, Integer pageSize, String startTime, String endTime, String result, String statusIssue, String imei, String supplierId, String companyId, String operator, Integer houseId, String bindingInfo) { JSONObject resultSet = new JSONObject(); InfluxDBClient client = InfluxDBClientFactory.create(url, token.toCharArray(), org); + try { // 1. 拼接公共过滤条件片段（总条数和分页查询共用） StringBuilder filterFragment = new StringBuilder(); filterFragment.append("from(bucket: \"devicelog\") "); - if (startTime != null && startTime != "" && endTime != null && endTime != "") { + if (startTime != null && !startTime.isEmpty() && endTime != null && !endTime.isEmpty()) { SimpleDateFormat oldFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); SimpleDateFormat newFormat = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'"); newFormat.setTimeZone(TimeZone.getTimeZone("UTC"));//时区转换 @@ -185,63 +220,52 @@ public class DeviceLogService { // 2. 拼接固定过滤条件（_measurement和_field，必选） filterFragment.append(" |> filter(fn: (r) => r[\"_measurement\"] == \"devicelog\")"); filterFragment.append(" |> filter(fn: (r) => r[\"_field\"] == \"logjson\")"); - // 3. 动态拼接imei过滤条件（仅当imei非空且非空白字符串时拼接） + // 3. 动态拼接imei过滤条件 if (imei != null && !imei.trim().isEmpty()) { filterFragment.append(String.format(" |> filter(fn: (r) => r[\"imei\"] == \"%s\")", imei.trim())); } - // 4. 动态拼接result过滤条件（仅当result非空且非空白字符串时拼接） + // 4. 动态拼接result过滤条件 if (result != null && !result.trim().isEmpty()) { filterFragment.append(String.format(" |> filter(fn: (r) => r[\"result\"] == \"%s\")", result.trim())); } if (statusIssue != null && !statusIssue.trim().isEmpty()) { filterFragment.append(String.format(" |> filter(fn: (r) => r[\"statusIssue\"] == \"%s\")", statusIssue.trim())); } - // 5. 动态拼接supplierId过滤条件（仅当supplierId非空且非空白字符串时拼接） + // 5. 动态拼接supplierId过滤条件 if (supplierId != null && !supplierId.trim().isEmpty()) { filterFragment.append(String.format(" |> filter(fn: (r) => r[\"supplierId\"] == \"%s\")", supplierId.trim())); } - // 6. 动态拼接companyId过滤条件（仅当companyId非空且非空白字符串时拼接） + // 6. 动态拼接companyId过滤条件 if (companyId != null && !companyId.trim().isEmpty()) { filterFragment.append(String.format(" |> filter(fn: (r) => r[\"companyId\"] == \"%s\")", companyId.trim())); } - // 7. 动态拼接operator过滤条件（仅当operator非空且非空白字符串时拼接） + // 7. 动态拼接operator过滤条件 if (operator != null && !operator.trim().isEmpty()) { filterFragment.append(String.format(" |> filter(fn: (r) => r[\"operator\"] == \"%s\")", operator.trim())); } - // 8. 动态拼接houseId过滤条件（仅当houseId非空且非空白字符串时拼接） + // 8. 动态拼接houseId过滤条件 if (houseId != null) { filterFragment.append(String.format(" |> filter(fn: (r) => r[\"houseId\"] == \"%s\")", houseId)); } - // ========== 核心修复：Java 8不支持"""，改用+拼接多行字符串 ========== - // 1. 拼接模糊过滤条件（Flux逻辑不变，仅改字符串写法） - // 3. 核心修复：belonginfo模糊查询（两步修复） + // 9. 拼接模糊过滤条件 if (bindingInfo != null && !bindingInfo.trim().isEmpty()) { filterFragment.append(String.format(" |> filter(fn: (r) => r[\"deviceBelongInfo\"] =~ /.*" + bindingInfo.trim() + ".*/)")); } -// if (bindingInfo != null && !bindingInfo.trim().isEmpty()) { -// // 核心：containsStr实现模糊匹配，先判断字段非空避免报错 -// filterFragment.append(String.format( -// " |> filter(fn: (r) => contains(value: r[\"deviceBelongInfo\"], set: [ \"%s\"]))", -// bindingInfo.trim() -// )); -// } - // 5. 拼接固定排序条件（按时间倒序） - // 2. 构建总条数查询子句（yield命名为total） + // 构建总条数查询子句 StringBuilder totalQuery = new StringBuilder(); - totalQuery.append(filterFragment); // 复用过滤条件 - totalQuery.append(" |> count(column: \"_value\")"); // 统计总条数 - totalQuery.append(" |> yield(name: \"total\")\n"); // 标记结果集为total + totalQuery.append(filterFragment); + totalQuery.append(" |> count(column: \"_value\")"); + totalQuery.append(" |> yield(name: \"total\")\n"); - // 3. 构建分页数据查询子句（yield命名为data） + // 构建分页数据查询子句 StringBuilder dataQuery = new StringBuilder(); - dataQuery.append(filterFragment); // 复用过滤条件 + dataQuery.append(filterFragment); dataQuery.append(" |> sort(columns:[\"_time\"], desc:true)"); // 处理分页参数默认值 int finalPageNum = pageNode != null ? pageNode : DEFAULT_PAGE_NUM; int finalPageSize = pageSize != null ? pageSize : DEFAULT_PAGE_SIZE; int offset = (finalPageNum - 1) * finalPageSize; -// dataQuery.append(" |> limit(n: " + finalPageSize + ", offset: " + offset + ")"); - dataQuery.append(" |> yield(name: \"data\")"); // 标记结果集为data + dataQuery.append(" |> yield(name: \"data\")"); System.out.println("查询数量语句：" + totalQuery.toString()); System.out.println("查询数据语句：" + dataQuery.toString()); //查询总条数 @@ -269,7 +293,6 @@ public class DeviceLogService { rerurnList.add(data); } } - client.close(); Collections.reverse(rerurnList); resultSet.put("count", totalCount); //对结果按照分页要求截取构造数据 @@ -278,17 +301,18 @@ public class DeviceLogService { } resultSet.put("list", rerurnList); return resultSet; + } finally { + client.close(); + } } /** * Flux正则转义（仅转义RE2引擎的特殊字符，避免\Q\E） - * Flux正则特殊字符：. * + ? | ( ) [ ] { } ^ $ \ */ private static String escapeFluxRegex(String keyword) { if (keyword == null || keyword.isEmpty()) { return ""; } - // 转义Flux正则的特殊字符（替换为\+字符） String[] specialChars = {"\\", ".", "*", "+", "?", "|", "(", ")", "[", "]", "{", "}", "^", "$"}; String escaped = keyword; for (String ch : specialChars) { @@ -301,17 +325,14 @@ public class DeviceLogService { * 转存设备命令下发日志数据 */ public void transferDeviceLogData(String searchImei, Integer companyId) { - //1. 查询所有的设备编号 List syncDataFlags = deviceInfoMapper.querySyncDeviceLogFlagInfo(searchImei); if (syncDataFlags != null && syncDataFlags.size() > 0) { - //2. 根据设备编号查询设备命令下发日志数据 for (SyncDataFlag syncDataFlag : syncDataFlags) { Boolean flag = false; String imei = syncDataFlag.getImei(); List deviceLogDataList = deviceInfoMapper.queryDeviceLogData(imei, companyId); if (deviceLogDataList != null && deviceLogDataList.size() > 0) { - DeviceBelongInfo deviceBelongInfo = deviceInfoMapper.queryDeviceBelongInfo(imei); - //3. 批量插入设备命令下发日志数据到influxdb中 + DeviceBelongInfo deviceBelongInfo = deviceInfoMapper.queryDeviceBelongInfo(imei); for (DeviceLogData deviceLogData : deviceLogDataList) { deviceLogData.setDeviceBelongInfo(deviceBelongInfo != null ? deviceBelongInfo.getDeviceBelongInfo() : null); deviceLogData.setHouseId(deviceBelongInfo != null ? deviceBelongInfo.getHouseId() : null); @@ -329,7 +350,6 @@ public class DeviceLogService { } } if (flag) { - //4. 更新同步设备命令下发日志数据状态 deviceInfoMapper.updateSyncDeviceLogFlagInfo(syncDataFlag.getId(), 1); } } @@ -340,44 +360,71 @@ public class DeviceLogService { * 更新设备命令下发日志数据 */ public void updateDeviceLogMqListener(DeviceLogData deviceLogData) { - //步骤1：查出ID对应的日志 InfluxDBClient client = InfluxDBClientFactory.create(url, token.toCharArray(), org); - // 1. 拼接公共过滤条件片段（总条数和分页查询共用） - StringBuilder filterFragment = new StringBuilder(); - filterFragment.append("from(bucket: \"devicelog\") "); - filterFragment.append("|> range(start: -360d)"); - // 2. 拼接固定过滤条件（_measurement和_field，必选） - filterFragment.append(" |> filter(fn: (r) => r[\"_measurement\"] == \"devicelog\")"); - filterFragment.append(" |> filter(fn: (r) => r[\"_field\"] == \"logjson\")"); - // 3. 动态拼接imei过滤条件（仅当imei非空且非空白字符串时拼接） - filterFragment.append(String.format(" |> filter(fn: (r) => r[\"id\"] == \"%s\")", deviceLogData.getId())); - // 3. 构建分页数据查询子句（yield命名为data） - StringBuilder dataQuery = new StringBuilder(); - dataQuery.append(filterFragment); // 复用过滤条件 - dataQuery.append(" |> sort(columns:[\"_time\"], desc:true)"); - dataQuery.append(" |> yield(name: \"data\")"); // 标记结果集为data - System.out.println("查询数据语句：" + dataQuery.toString()); - //查询数据结果 - List dataResults = client.getQueryApi().query(dataQuery.toString()); - for (FluxTable fluxTable : dataResults) { - List records = fluxTable.getRecords(); - for (FluxRecord fluxRecord : records) { - String value = (String) fluxRecord.getValueByKey("_value"); - DeviceLogData data = JSONObject.parseObject(value, DeviceLogData.class); - //构造数据 - DeviceLogData newData = new DeviceLogData(); - BeanUtils.copyProperties(data, newData); - newData.setId(deviceLogData.getId()); - newData.setFeedbackValue(deviceLogData.getFeedbackValue()); - newData.setStatusIssue(deviceLogData.getStatusIssue()); - //步骤2：删除该日志 - //删除设备命令下发日志数据 - deleteDeviceLog(deviceLogData.getId()); - //步骤3：更新该日志 - //保存设备命令下发日志数据到influxdb中 - saveDeviceLogToInfluxdb(newData); + try { + StringBuilder filterFragment = new StringBuilder(); + filterFragment.append("from(bucket: \"devicelog\") "); + filterFragment.append("|> range(start: -360d)"); + filterFragment.append(" |> filter(fn: (r) => r[\"_measurement\"] == \"devicelog\")"); + filterFragment.append(" |> filter(fn: (r) => r[\"_field\"] == \"logjson\")"); + filterFragment.append(String.format(" |> filter(fn: (r) => r[\"id\"] == \"%s\")", deviceLogData.getId())); + StringBuilder dataQuery = new StringBuilder(); + dataQuery.append(filterFragment); + dataQuery.append(" |> sort(columns:[\"_time\"], desc:true)"); + dataQuery.append(" |> yield(name: \"data\")"); + System.out.println("查询数据语句：" + dataQuery.toString()); + List dataResults = client.getQueryApi().query(dataQuery.toString()); + for (FluxTable fluxTable : dataResults) { + List records = fluxTable.getRecords(); + for (FluxRecord fluxRecord : records) { + String value = (String) fluxRecord.getValueByKey("_value"); + DeviceLogData data = JSONObject.parseObject(value, DeviceLogData.class); + DeviceLogData newData = new DeviceLogData(); + BeanUtils.copyProperties(data, newData); + newData.setId(deviceLogData.getId()); + newData.setFeedbackValue(deviceLogData.getFeedbackValue()); + newData.setStatusIssue(deviceLogData.getStatusIssue()); + deleteDeviceLog(deviceLogData.getId()); + saveDeviceLogToInfluxdb(newData); + } + } + } finally { + client.close(); + } + } + + /** + * 检查异常是否为 InfluxDB 超时、连接失败或 shard 损坏异常 + */ + private boolean isInfluxDBTimeoutException(Exception e) { + String message = e.getMessage(); + if (message != null && (message.contains("not attempting to open shard") || + message.contains("short buffer") || + message.contains("InternalServerErrorException"))) { + return true; + } + if (message != null && (message.contains("Failed to connect") || + message.contains("ConnectException") || + message.contains("Connection refused") || + message.contains("Connection timed out") || + message.contains("connect timed out"))) { + return true; + } + Throwable cause = e.getCause(); + Set visitedCauses = new HashSet<>(); + while (cause != null && visitedCauses.add(cause)) { + if (cause instanceof java.net.SocketTimeoutException || + cause instanceof java.net.ConnectException) { + return true; + } + if (cause.getMessage() != null && + (cause.getMessage().contains("Read timed out") || + cause.getMessage().contains("connect timed out") || + cause.getMessage().contains("Connection timed out"))) { + return true; } + cause = cause.getCause(); } - client.close(); + return false; } } diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties index 7ef22a9..4aef16d 100644 --- a/src/main/resources/application.properties +++ b/src/main/resources/application.properties @@ -8,12 +8,41 @@ spring.rabbitmq.username=topsail spring.rabbitmq.password=topsail spring.rabbitmq.virtualHost=/ spring.rabbitmq.listener.simple.acknowledge-mode=manual -spring.rabbitmq.listener.simple.prefetch=10 +spring.rabbitmq.listener.simple.prefetch=100 +spring.rabbitmq.listener.simple.concurrency=5 +spring.rabbitmq.listener.simple.max-concurrency=10 -spring.datasource.url=jdbc:mysql://rm-2ze77qng1ddlfur9g4o.mysql.rds.aliyuncs.com:3306/zhrl?useSSL=false&serverTimezone=UTC&useUnicode=true&characterEncoding=UTF-8&allowMultiQueries=true&zeroDateTimeBehavior=convertToNull +# RabbitMQ Management API 配置（用于获取队列列表） +rabbitmq.management.url=http://182.92.218.150:15672 +rabbitmq.management.username=topsail +rabbitmq.management.password=topsail +rabbitmq.management.vhost=/ + +spring.datasource.url=jdbc:mysql://rm-2ze77qng1ddlfur9g4o.mysql.rds.aliyuncs.com:3306/zhrl?useSSL=false&serverTimezone=UTC&useUnicode=true&characterEncoding=UTF-8&allowMultiQueries=true&zeroDateTimeBehavior=convertToNull&autoReconnect=true&failOverReadOnly=false&maxReconnects=3&initialTimeout=2&connectTimeout=60000&socketTimeout=60000 spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver spring.datasource.username=topsail spring.datasource.password=Topsail2020 + +# HikariCP 连接池优化配置 +# 最小空闲连接数 +spring.datasource.hikari.minimum-idle=5 +# 最大连接池大小（根据并发需求调整） +spring.datasource.hikari.maximum-pool-size=20 +# 连接超时时间（毫秒） +spring.datasource.hikari.connection-timeout=30000 +# 连接最大生命周期（毫秒），设置为 2 分钟，必须小于 MySQL 的 wait_timeout +spring.datasource.hikari.max-lifetime=120000 +# 空闲连接超时时间（毫秒），设置为 1 分钟 +spring.datasource.hikari.idle-timeout=60000 +# 连接测试查询 +spring.datasource.hikari.connection-test-query=SELECT 1 +# 自动提交 +spring.datasource.hikari.auto-commit=true +# 泄漏检测阈值（毫秒），超过此时间的连接会被记录警告 +spring.datasource.hikari.leak-detection-threshold=60000 +# 保持连接活跃，每 30 秒发送一次心跳 +spring.datasource.hikari.keepalive-time=30000 + ##正式环境influxdb配置 shengdilan.influxdb.token=0rg4n4KBC6x65pljf-OzaqvXrRCJGKQxl_ZGSSijdTRKNuVgbeTDMf5UKIHZPYHKjCHVrnKKNOu9hVVVNUCaZw== shengdilan.influxdb.url=http://113.137.28.150:8086 @@ -30,4 +59,17 @@ shengdilan.influxdb.oldorg=topsail ##旧influxdb配置参数 #shengdilan.influxdb.oldtoken=C2sfXsMC475aTtin7HbRkUXa9tEZTUU0S928ZdPzFktcFW8gZD_zY8-hKhgPxkLLodVS4YcsL3RcwgsJWYlURw== #shengdilan.influxdb.oldurl=http://192.168.139.128:8086 -#shengdilan.influxdb.oldorg=shengdilan \ No newline at end of file +#shengdilan.influxdb.oldorg=shengdilan + +# ==================== MySQL SQL 打印配置 ==================== +# 方式1：MyBatis 日志打印（推荐） +# 打印所有 Mapper 接口的 SQL 语句 +logging.level.com.topsail.influxdb.mapper=DEBUG + +# 方式2：Spring JDBC 日志打印（可选） +# 打印数据源相关的 SQL +# logging.level.org.springframework.jdbc.core=DEBUG +# logging.level.org.springframework.jdbc.datasource=DEBUG + +# 方式3：HikariCP 连接池日志（可选） +# logging.level.com.zaxxer.hikari=DEBUG