package com.ard.web.controller.sample.media; import com.ard.oss.model.CredentialsDTO; import com.ard.oss.service.impl.OssServiceContext; import io.swagger.annotations.Api; import io.swagger.annotations.ApiOperation; import lombok.extern.slf4j.Slf4j; import okhttp3.*; import okio.BufferedSink; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; import javax.net.ssl.*; import javax.servlet.http.HttpServletRequest; import javax.servlet.http.HttpServletResponse; import java.io.ByteArrayOutputStream; import java.io.IOException; import java.io.InputStream; import java.security.SecureRandom; import java.util.Enumeration; import java.util.concurrent.TimeUnit; @Api(tags = "机场上传minio") @Slf4j @RestController @RequestMapping("/minioTransfer") public class MinioStsProxyController { // ==================== 常量配置(适配大文件) ==================== private static final String PROXY_PREFIX = "/minioTransfer"; private static final int BUFFER_SIZE = 65536; // 64K缓冲区,提升大文件传输效率 private static final long MAX_MEMORY_SIZE = 1024 * 1024 * 100; // 100MB,超过则用流式 private static final String X_AMZ_SECURITY_TOKEN = "X-Amz-Security-Token"; private static final String TRANSFER_ENCODING = "Transfer-Encoding"; private static final String CONTENT_LENGTH = "Content-Length"; private static final String HOST = "Host"; // 配置文件注入(大幅提升超时时间) @Value("${oss.endpoint}") private String minioEndpoint; private static final Integer connectTimeout=30; private static final Integer readWriteTimeout=1800; @Autowired private OssServiceContext ossService; // OkHttp 单例客户端(全局复用) private static OkHttpClient httpClient; // 静态初始化,强制TLS 1.2,修复SSL兼容问题 static { createOkHttpClient(); } /** * 重构OkHttp客户端:强制TLS 1.2 + 超长超时 + 连接保活 */ private static void createOkHttpClient() { try { // 1. 强制使用TLS 1.2,解决SSL版本不兼容导致的连接重置 SSLContext sslContext = SSLContext.getInstance("TLSv1.2"); TrustManager[] trustAllCerts = new TrustManager[]{ new X509TrustManager() { @Override public void checkClientTrusted(java.security.cert.X509Certificate[] certs, String authType) {} @Override public void checkServerTrusted(java.security.cert.X509Certificate[] certs, String authType) {} @Override public java.security.cert.X509Certificate[] getAcceptedIssuers() { return new java.security.cert.X509Certificate[0]; } } }; sslContext.init(null, trustAllCerts, new SecureRandom()); SSLSocketFactory sslSocketFactory = sslContext.getSocketFactory(); // 2. 配置连接工厂,强制TLS 1.2 ConnectionSpec connectionSpec = new ConnectionSpec.Builder(ConnectionSpec.MODERN_TLS) .tlsVersions(TlsVersion.TLS_1_2) .build(); // 3. 构建客户端,增加连接保活和重试 httpClient = new OkHttpClient.Builder() .sslSocketFactory(sslSocketFactory, (X509TrustManager) trustAllCerts[0]) .hostnameVerifier((hostname, session) -> true) .connectionSpecs(java.util.Collections.singletonList(connectionSpec)) // 强制TLS 1.2 .connectTimeout(connectTimeout, TimeUnit.SECONDS) .readTimeout(readWriteTimeout, TimeUnit.SECONDS) .writeTimeout(readWriteTimeout, TimeUnit.SECONDS) .connectionPool(new ConnectionPool(20, 10, TimeUnit.MINUTES)) // 更大的连接池 .retryOnConnectionFailure(true) // 开启连接失败重试 .pingInterval(30, TimeUnit.SECONDS) // TCP保活,防止连接被断开 .build(); } catch (Exception e) { log.error("OkHttp初始化失败,使用默认客户端", e); httpClient = new OkHttpClient(); } } @ApiOperation("MinIO STS临时凭证通用代理") @RequestMapping("/**") public void proxyRequest(HttpServletRequest request, HttpServletResponse response) { String method = request.getMethod(); String requestUri = request.getRequestURI(); String queryString = request.getQueryString(); log.info("STS代理请求开始 | 方法:{} | URI:{} | 参数:{} | 内容长度:{}", method, requestUri, queryString, request.getContentLengthLong()); // 构造目标地址 String targetUrl = buildTargetUrl(requestUri, queryString); if (targetUrl == null) { sendError(response, HttpServletResponse.SC_BAD_REQUEST, "请求路径非法"); return; } // 获取STS凭证 CredentialsDTO credentials = getCredentials(); if (credentials == null) { sendError(response, HttpServletResponse.SC_FORBIDDEN, "STS凭证无效"); return; } try { // 构建请求(区分大/小文件上传方式) Request okRequest = buildRequest(request, method, targetUrl, credentials); // 执行并转发响应 forwardResponse(okRequest, response); log.info("STS代理请求成功 | 目标地址:{}", targetUrl); } catch (Exception e) { log.error("STS代理请求执行异常", e); sendError(response, HttpServletResponse.SC_INTERNAL_SERVER_ERROR, "代理服务异常:" + e.getMessage()); } } /** * 构造MinIO请求地址 */ private String buildTargetUrl(String requestUri, String queryString) { try { String targetPath = requestUri.substring(PROXY_PREFIX.length()); StringBuilder urlBuilder = new StringBuilder(minioEndpoint).append(targetPath); if (queryString != null && !queryString.isBlank()) { urlBuilder.append("?").append(queryString); } return urlBuilder.toString(); } catch (Exception e) { log.error("目标地址构造失败", e); return null; } } /** * 获取并校验STS临时凭证 */ private CredentialsDTO getCredentials() { try { CredentialsDTO dto = ossService.getCredentials(); if (dto == null || dto.getAccessKeyId() == null || dto.getSecurityToken() == null) { log.error("STS凭证核心字段缺失"); return null; } log.info("STS凭证获取成功 | AK前缀:{}", dto.getAccessKeyId()); //.substring(0, 4) + "***" return dto; } catch (Exception e) { log.error("获取STS凭证异常", e); return null; } } /** * 构建OkHttp请求:核心优化大文件上传逻辑 */ private Request buildRequest(HttpServletRequest request, String method, String url, CredentialsDTO credentials) throws IOException { Request.Builder builder = new Request.Builder().url(url); // 透传请求头:仅过滤Host,保留Content-Length Enumeration headerNames = request.getHeaderNames(); while (headerNames.hasMoreElements()) { String name = headerNames.nextElement(); if (name.equalsIgnoreCase(HOST)) { continue; } builder.addHeader(name, request.getHeader(name)); } // 强制添加STS安全令牌 builder.addHeader(X_AMZ_SECURITY_TOKEN, credentials.getSecurityToken()); // 构造请求体:区分GET/HEAD和其他方法 RequestBody body = null; if (!"GET".equalsIgnoreCase(method) && !"HEAD".equalsIgnoreCase(method)) { long contentLength = request.getContentLengthLong(); InputStream inputStream = request.getInputStream(); // 策略1:小文件(≤100MB)→ 预读取到字节数组,避免流式传输的连接不稳定 if (contentLength > 0 && contentLength <= MAX_MEMORY_SIZE) { log.info("小文件上传:预读取到内存,长度:{}MB", contentLength / 1024 / 1024); byte[] content = readInputStreamToBytes(inputStream); MediaType mediaType = request.getContentType() != null ? MediaType.parse(request.getContentType()) : MediaType.parse("application/octet-stream"); body = RequestBody.create(content, mediaType); } // 策略2:大文件/未知长度 → 流式上传,优化超时和缓冲区 else { log.info("大文件/未知长度上传:使用流式传输"); body = new RequestBody() { @Override public MediaType contentType() { String contentType = request.getContentType(); return contentType != null ? MediaType.parse(contentType) : MediaType.parse("application/octet-stream"); } @Override public long contentLength() { return contentLength; // 返回原始长度,避免chunked编码 } @Override public void writeTo(BufferedSink sink) throws IOException { // 优化:增加异常捕获和重试逻辑 byte[] buffer = new byte[BUFFER_SIZE]; int len; long totalWritten = 0; try (InputStream in = inputStream) { while ((len = in.read(buffer)) != -1) { sink.write(buffer, 0, len); totalWritten += len; // 每写100MB刷新一次,防止缓冲区溢出 if (totalWritten % (1024 * 1024 * 100) == 0) { sink.flush(); log.info("已上传:{}MB", totalWritten / 1024 / 1024); } } sink.flush(); // 最终刷新 log.info("流式上传完成,总写入:{}MB", totalWritten / 1024 / 1024); } catch (IOException e) { log.error("流式上传失败,已写入:{}MB", totalWritten / 1024 / 1024, e); throw e; // 抛出异常让上层处理 } } }; } } builder.method(method, body); return builder.build(); } /** * 工具方法:读取InputStream到字节数组(小文件专用) */ private byte[] readInputStreamToBytes(InputStream inputStream) throws IOException { ByteArrayOutputStream bos = new ByteArrayOutputStream(); byte[] buffer = new byte[BUFFER_SIZE]; int len; while ((len = inputStream.read(buffer)) != -1) { bos.write(buffer, 0, len); } bos.flush(); byte[] result = bos.toByteArray(); bos.close(); return result; } /** * 执行请求并转发响应:增加响应体处理的容错 */ private void forwardResponse(Request request, HttpServletResponse response) throws IOException { try (Response okResponse = httpClient.newCall(request).execute()) { // 设置响应状态码 response.setStatus(okResponse.code()); // 透传响应头 Headers headers = okResponse.headers(); for (String name : headers.names()) { if (TRANSFER_ENCODING.equalsIgnoreCase(name)) { continue; } response.setHeader(name, headers.get(name)); } // 透传响应体:增加空判断和流关闭容错 ResponseBody responseBody = okResponse.body(); if (responseBody != null) { try (InputStream in = responseBody.byteStream()) { byte[] buffer = new byte[BUFFER_SIZE]; int len; while ((len = in.read(buffer)) != -1) { response.getOutputStream().write(buffer, 0, len); } response.getOutputStream().flush(); } catch (IOException e) { log.warn("响应体转发失败(客户端可能已断开连接)", e); } } } } /** * 统一错误响应 */ private void sendError(HttpServletResponse response, int code, String msg) { try { response.sendError(code, msg); } catch (IOException e) { log.error("响应错误信息失败", e); } } }