195 lines
8.2 KiB
Java
195 lines
8.2 KiB
Java
package com.iot.gateway.common;
|
|
|
|
import com.fasterxml.jackson.databind.JsonNode;
|
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
|
import java.net.ConnectException;
|
|
import java.net.SocketTimeoutException;
|
|
import java.net.URI;
|
|
import java.net.URLEncoder;
|
|
import java.net.http.HttpClient;
|
|
import java.net.http.HttpConnectTimeoutException;
|
|
import java.net.http.HttpRequest;
|
|
import java.net.http.HttpResponse;
|
|
import java.net.http.HttpTimeoutException;
|
|
import java.nio.charset.StandardCharsets;
|
|
import java.time.Duration;
|
|
import java.util.HashMap;
|
|
import java.util.Map;
|
|
import lombok.extern.slf4j.Slf4j;
|
|
import org.springframework.beans.factory.annotation.Value;
|
|
import org.springframework.stereotype.Component;
|
|
|
|
/**
|
|
* 网关与核心业务后端 (IOT-backend) 的专线 HTTP 桥接客户端
|
|
* <p>
|
|
* 具备 URL 安全编码、错误细粒度分类、幂等 GET 有限重试与租户强制隔离。
|
|
*/
|
|
@Slf4j
|
|
@Component
|
|
public class BackendInternalClient {
|
|
|
|
@Value("${gateway.backend.base-url:http://localhost:8080}")
|
|
private String backendBaseUrl;
|
|
|
|
@Value("${gateway.backend.internal-secret:}")
|
|
private String internalSecret;
|
|
|
|
private final HttpClient httpClient = HttpClient.newBuilder()
|
|
.connectTimeout(Duration.ofSeconds(4))
|
|
.build();
|
|
|
|
private final ObjectMapper objectMapper = new ObjectMapper();
|
|
|
|
/**
|
|
* 内部专线 GET 请求 (支持参数 URL 编码、错误细粒度分类与网络抖动单次有限重试)
|
|
*/
|
|
public JsonNode get(String path, Long tenantId, Map<String, String> queryParams) {
|
|
if (tenantId == null) {
|
|
throw new BizException(401, "专线请求缺少租户上下文，严禁跨租户或无租户调用");
|
|
}
|
|
|
|
String fullUrl = buildGetUrl(path, queryParams);
|
|
int maxAttempts = 2; // 最多尝试2次（初次 + 1次重试）
|
|
|
|
for (int attempt = 1; attempt <= maxAttempts; attempt++) {
|
|
try {
|
|
HttpRequest request = HttpRequest.newBuilder()
|
|
.uri(URI.create(fullUrl))
|
|
.header("X-Internal-Secret", internalSecret != null ? internalSecret : "")
|
|
.header("X-Tenant-Id", String.valueOf(tenantId))
|
|
.header("Content-Type", "application/json")
|
|
.timeout(Duration.ofSeconds(5))
|
|
.GET()
|
|
.build();
|
|
|
|
HttpResponse<String> resp = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
|
|
return handleResponse(resp, fullUrl);
|
|
|
|
} catch (HttpTimeoutException | SocketTimeoutException e) {
|
|
if (attempt < maxAttempts) {
|
|
log.warn("GET 请求超时，正在进行瞬时网络重试 (attempt={}): url={}", attempt, fullUrl);
|
|
backoff(200);
|
|
continue;
|
|
}
|
|
log.error("核心后端 GET 请求超时: url={}, tenantId={}, err={}", fullUrl, tenantId, e.getMessage());
|
|
throw new BizException(504, "调用核心后端业务超时，服务可能处于高负荷: " + e.getMessage());
|
|
|
|
} catch (ConnectException e) {
|
|
if (attempt < maxAttempts) {
|
|
log.warn("GET 连接异常，正在尝试重连 (attempt={}): url={}", attempt, fullUrl);
|
|
backoff(200);
|
|
continue;
|
|
}
|
|
log.error("核心后端连接拒绝: url={}, tenantId={}, err={}", fullUrl, tenantId, e.getMessage());
|
|
throw new BizException(503, "核心后端服务不可达，请检查后端运行状态");
|
|
|
|
} catch (BizException be) {
|
|
throw be;
|
|
|
|
} catch (Exception e) {
|
|
log.error("调用核心后端 GET 异常: url={}, tenantId={}, err={}", fullUrl, tenantId, e.getMessage());
|
|
throw new BizException(502, "核心物联后端通信异常: " + e.getMessage());
|
|
}
|
|
}
|
|
|
|
throw new BizException(504, "调用核心后端业务多次超时");
|
|
}
|
|
|
|
/**
|
|
* 内部专线 POST 请求 (设备反写与业务提交严禁重试，避免指令重复执行)
|
|
*/
|
|
public JsonNode post(String path, Long tenantId, Map<String, Object> body) {
|
|
if (tenantId == null) {
|
|
throw new BizException(401, "专线请求缺少租户上下文，严禁跨租户或无租户调用");
|
|
}
|
|
|
|
String fullUrl = backendBaseUrl + path;
|
|
try {
|
|
String jsonStr = objectMapper.writeValueAsString(body != null ? body : new HashMap<>());
|
|
HttpRequest request = HttpRequest.newBuilder()
|
|
.uri(URI.create(fullUrl))
|
|
.header("X-Internal-Secret", internalSecret != null ? internalSecret : "")
|
|
.header("X-Tenant-Id", String.valueOf(tenantId))
|
|
.header("Content-Type", "application/json")
|
|
.timeout(Duration.ofSeconds(6))
|
|
.POST(HttpRequest.BodyPublishers.ofString(jsonStr))
|
|
.build();
|
|
|
|
HttpResponse<String> resp = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
|
|
return handleResponse(resp, fullUrl);
|
|
|
|
} catch (HttpTimeoutException | SocketTimeoutException e) {
|
|
log.error("核心后端 POST 请求超时 (未重试，防止指令重放): url={}, tenantId={}", fullUrl, tenantId);
|
|
throw new BizException(504, "核心控制通道通信超时，无法确认物理设备指令下发状态");
|
|
|
|
} catch (ConnectException e) {
|
|
log.error("核心后端连接失败: url={}, tenantId={}, err={}", fullUrl, tenantId, e.getMessage());
|
|
throw new BizException(503, "核心控制网关服务不可达");
|
|
|
|
} catch (BizException be) {
|
|
throw be;
|
|
|
|
} catch (Exception e) {
|
|
log.error("核心后端 POST 调用异常: url={}, tenantId={}, err={}", fullUrl, tenantId, e.getMessage());
|
|
throw new BizException(502, "控制指令网关通信失败: " + e.getMessage());
|
|
}
|
|
}
|
|
|
|
private JsonNode handleResponse(HttpResponse<String> resp, String fullUrl) {
|
|
int status = resp.statusCode();
|
|
String body = resp.body();
|
|
|
|
if (status == 200) {
|
|
try {
|
|
return objectMapper.readTree(body);
|
|
} catch (Exception e) {
|
|
log.error("核心后端响应非标准 JSON: body={}", body);
|
|
throw new BizException(502, "核心后端返回格式错误");
|
|
}
|
|
}
|
|
|
|
if (status == 401 || status == 403) {
|
|
log.error("专线内部鉴权失败: status={}, body={}, url={}", status, body, fullUrl);
|
|
throw new BizException(500, "内部通信专线密钥配置不匹配或租户无权访问");
|
|
}
|
|
|
|
if (status == 404) {
|
|
throw new BizException(404, "核心后端未找到目标接口: " + fullUrl);
|
|
}
|
|
|
|
log.warn("核心后端返回错误状态码: status={}, body={}, url={}", status, body, fullUrl);
|
|
try {
|
|
JsonNode errJson = objectMapper.readTree(body);
|
|
if (errJson.has("message")) {
|
|
throw new BizException(502, errJson.get("message").asText());
|
|
}
|
|
} catch (BizException be) {
|
|
throw be;
|
|
} catch (Exception ignored) {}
|
|
|
|
throw new BizException(502, "核心后端服务返回异常代码: HTTP " + status);
|
|
}
|
|
|
|
private String buildGetUrl(String path, Map<String, String> queryParams) {
|
|
StringBuilder urlBuilder = new StringBuilder(backendBaseUrl).append(path);
|
|
if (queryParams != null && !queryParams.isEmpty()) {
|
|
urlBuilder.append("?");
|
|
for (Map.Entry<String, String> entry : queryParams.entrySet()) {
|
|
if (entry.getValue() != null) {
|
|
String k = URLEncoder.encode(entry.getKey(), StandardCharsets.UTF_8);
|
|
String v = URLEncoder.encode(entry.getValue(), StandardCharsets.UTF_8);
|
|
urlBuilder.append(k).append("=").append(v).append("&");
|
|
}
|
|
}
|
|
urlBuilder.deleteCharAt(urlBuilder.length() - 1);
|
|
}
|
|
return urlBuilder.toString();
|
|
}
|
|
|
|
private void backoff(long ms) {
|
|
try {
|
|
Thread.sleep(ms);
|
|
} catch (InterruptedException ignored) {}
|
|
}
|
|
}
|