bgy
1 month ago
7 changed files with 1424 additions and 262 deletions
Unified View
Diff Options
-
+9 -1pom.xml
-
+409 -0src/main/java/com/topsail/influxdb/config/InfluxDBConfig.java
-
+227 -14src/main/java/com/topsail/influxdb/rabbitmq/AmqpListener.java
-
+221 -12src/main/java/com/topsail/influxdb/rabbitmq/service/AmqpService.java
-
+378 -144src/main/java/com/topsail/influxdb/service/DeviceDataService.java
-
+135 -88src/main/java/com/topsail/influxdb/service/DeviceLogService.java
-
+45 -3src/main/resources/application.properties
+ 9
- 1
pom.xml
View File
+ 409
- 0
src/main/java/com/topsail/influxdb/config/InfluxDBConfig.java
View File
| @ -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<Runnable> 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()); | |||||
| } | |||||
| } | |||||
| } | |||||
| } | |||||
+ 227
- 14
src/main/java/com/topsail/influxdb/rabbitmq/AmqpListener.java
View File
+ 221
- 12
src/main/java/com/topsail/influxdb/rabbitmq/service/AmqpService.java
View File
| @ -1,46 +1,255 @@ | |||||
| package com.topsail.influxdb.rabbitmq.service; | 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.Logger; | ||||
| import org.slf4j.LoggerFactory; | import org.slf4j.LoggerFactory; | ||||
| import org.springframework.amqp.core.AmqpAdmin; | import org.springframework.amqp.core.AmqpAdmin; | ||||
| import org.springframework.amqp.core.AmqpTemplate; | import org.springframework.amqp.core.AmqpTemplate; | ||||
| import org.springframework.amqp.core.Message; | 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.Autowired; | ||||
| import org.springframework.beans.factory.annotation.Value; | |||||
| import org.springframework.stereotype.Service; | 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 | @Service | ||||
| public class AmqpService { | public class AmqpService { | ||||
| public static final Logger LOG = LoggerFactory.getLogger(AmqpService.class); | public static final Logger LOG = LoggerFactory.getLogger(AmqpService.class); | ||||
| private final AmqpAdmin amqpAdmin; | private final AmqpAdmin amqpAdmin; | ||||
| private final AmqpTemplate amqpTemplate; | 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 | @Autowired | ||||
| public AmqpService(AmqpAdmin amqpAdmin, AmqpTemplate amqpTemplate) { | |||||
| public AmqpService(AmqpAdmin amqpAdmin, AmqpTemplate amqpTemplate, ConnectionFactory connectionFactory) { | |||||
| this.amqpAdmin = amqpAdmin; | this.amqpAdmin = amqpAdmin; | ||||
| this.amqpTemplate = amqpTemplate; | 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); | Message msg = this.amqpTemplate.receive(queue); | ||||
| if(msg!=null){ | |||||
| this.amqpTemplate.send(queue,msg); | |||||
| if (msg != null) { | |||||
| this.amqpTemplate.send(queue, msg); | |||||
| return false; | return false; | ||||
| }else{ | |||||
| } else { | |||||
| return true; | return true; | ||||
| } | } | ||||
| }catch (Exception e){ | |||||
| } catch (Exception e) { | |||||
| return true; | 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<String> getAllQueueNames() { | |||||
| List<String> 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; | |||||
| } | |||||
| } | } | ||||