客戶端 → 網(wǎng)關(guān) → 讀取請求體 → 解析參數(shù) → 調(diào)用驗(yàn)證服務(wù) → 轉(zhuǎn)發(fā)請求

// 傳統(tǒng)方案：3次內(nèi)存拷貝
byte[] heapCopy = readFromSocket();           // 1. 網(wǎng)卡→堆內(nèi)存
Map<String, Object> parsed = parse(heapCopy); // 2. 堆內(nèi)存→Java對象
byte[] jsonBytes = serialize(parsed);         // 3. Java對象→字節(jié)數(shù)組

客戶端
  ↓
┌─────────────────────────────────┐
│   Spring Cloud Gateway          │
│  ┌─────────────────────────────┐ │
│  │   1. 接收請求               │ │
│  │   2. 提取元數(shù)據(jù)             │ │
│  │   3. 零拷貝創(chuàng)建兩份視圖     │ │
│  └─────────────────────────────┘ │
└──────────────┬──────────────────┘
               │
    ┌──────────┼──────────┐
    ↓          ↓          ↓
驗(yàn)證服務(wù)      下游服務(wù)    監(jiān)控服務(wù)
（讀取視圖）  （讀取視圖） （元數(shù)據(jù)）

// 引用計(jì)數(shù)的正確使用模式
public class SafeReferenceCounting {
    
    public void process(ByteBuf original) {
        // 初始狀態(tài)：refCnt = 1
        
        // 創(chuàng)建兩個視圖
        ByteBuf view1 = original.duplicate().retain();  // refCnt = 2
        ByteBuf view2 = original.duplicate().retain();  // refCnt = 3
        
        try {
            // 并行處理兩個視圖
            processView1(view1);
            processView2(view2);
        } finally {
            // 必須釋放視圖
            view1.release();  // refCnt = 2
            view2.release();  // refCnt = 1
            
            // 注意：不釋放original，由框架管理
        }
    }
}

@Component
public class SharedBufferManager {
    
    /**
     * 創(chuàng)建可共享的緩沖區(qū)
     */
    public SharedBuffer wrap(DataBuffer buffer) {
        if (buffer instanceof NettyDataBuffer) {
            NettyDataBuffer nettyBuffer = (NettyDataBuffer) buffer;
            ByteBuf byteBuf = nettyBuffer.getNativeBuffer();
            
            // 增加引用計(jì)數(shù)
            byteBuf.retain();
            
            return new NettySharedBuffer(byteBuf, nettyBuffer.getDataBufferFactory());
        }
        
        // 非Netty緩沖區(qū)，回退到拷貝
        return new HeapSharedBuffer(buffer);
    }
    
    /**
     * 共享緩沖區(qū)接口
     */
    public interface SharedBuffer {
        DataBuffer createView();
        void releaseView(DataBuffer view);
        void close();
    }
}

@Component
@Order(-1)
public class ZeroCopyValidationFilter implements GlobalFilter {
    
    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        ServerHttpRequest request = exchange.getRequest();
        
        // 1. 判斷是否適用零拷貝
        if (!shouldUseZeroCopy(request)) {
            return chain.filter(exchange);
        }
        
        // 2. 合并請求體
        return DataBufferUtils.join(request.getBody())
            .flatMap(originalBuffer -> {
                // 3. 創(chuàng)建共享緩沖區(qū)
                try (SharedBuffer sharedBuffer = bufferManager.wrap(originalBuffer)) {
                    
                    // 4. 創(chuàng)建兩個視圖
                    DataBuffer validationView = sharedBuffer.createView();
                    DataBuffer forwardView = sharedBuffer.createView();
                    
                    // 5. 并行處理
                    return Mono.zip(
                        validate(validationView, exchange)
                            .doFinally(s -> sharedBuffer.releaseView(validationView)),
                        forward(forwardView, exchange)
                            .doFinally(s -> sharedBuffer.releaseView(forwardView))
                    ).flatMap(tuple -> {
                        boolean isValid = tuple.getT1();
                        if (isValid) {
                            return Mono.empty(); // 驗(yàn)證通過，請求已轉(zhuǎn)發(fā)
                        } else {
                            exchange.getResponse()
                                .setStatusCode(HttpStatus.UNAUTHORIZED);
                            return exchange.getResponse().setComplete();
                        }
                    });
                }
            });
    }
    
    private Mono<Boolean> validate(DataBuffer buffer, ServerWebExchange exchange) {
        // 提取元數(shù)據(jù)（不包含請求體）
        Map<String, String> metadata = extractMetadata(exchange);
        
        return webClient.post()
            .uri("http://validation-service/validate")
            .header("X-Request-Metadata", encodeMetadata(metadata))
            .contentType(MediaType.APPLICATION_OCTET_STREAM)
            .body(BodyInserters.fromDataBuffers(Flux.just(buffer)))
            .retrieve()
            .bodyToMono(ValidationResult.class)
            .map(ValidationResult::isValid)
            .timeout(Duration.ofMillis(500))
            .onErrorReturn(false);
    }
    
    private Mono<Void> forward(DataBuffer buffer, ServerWebExchange exchange) {
        ServerHttpRequest request = exchange.getRequest();
        
        return WebClient.create()
            .method(request.getMethod())
            .uri(request.getURI())
            .headers(headers -> headers.addAll(request.getHeaders()))
            .body(BodyInserters.fromDataBuffers(Flux.just(buffer)))
            .exchangeToMono(clientResponse -> {
                ServerHttpResponse response = exchange.getResponse();
                response.setStatusCode(clientResponse.statusCode());
                response.getHeaders()
                    .putAll(clientResponse.headers().asHttpHeaders());
                return response.writeWith(
                    clientResponse.bodyToFlux(DataBuffer.class)
                );
            });
    }
}

spring:
  cloud:
    gateway:
      httpclient:
        pool:
          max-connections: 1000
          max-idle-time: 60s
server:
  netty:
    use-native-transport: true
gateway:
  zerocopy:
    enabled: true
    max-request-size: 10MB
    content-types:
      - application/json
      - application/x-www-form-urlencoded
    timeout:
      validation: 500ms
      forward: 30s

@Configuration
public class NettyConfiguration {
    @Bean
    public NettyServerCustomizer nettyServerCustomizer() {
        return httpServer -> httpServer
            .tcpConfiguration(tcpServer -> tcpServer
                .selectorOption(ChannelOption.ALLOCATOR, 
                    PooledByteBufAllocator.DEFAULT)
                .selectorOption(ChannelOption.SO_BACKLOG, 10000)
            );
    }
    @Bean
    public HttpClient httpClient() {
        return HttpClient.create()
            .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
            .responseTimeout(Duration.ofSeconds(30));
    }
}

@Component
public class ZeroCopyMetrics {
    
    // 關(guān)鍵性能指標(biāo)
    private final Counter zeroCopyRequests = Counter.builder("gateway.zerocopy.requests")
        .description("零拷貝請求數(shù)量")
        .register(meterRegistry);
    
    private final Timer zeroCopyLatency = Timer.builder("gateway.zerocopy.latency")
        .description("零拷貝處理延遲")
        .register(meterRegistry);
    
    // 內(nèi)存使用指標(biāo)
    private final Gauge directMemoryUsage = Gauge.builder("gateway.memory.direct")
        .description("直接內(nèi)存使用量")
        .register(meterRegistry);
    
    // 記錄請求處理
    public void recordRequest(long size, long latency) {
        zeroCopyRequests.increment();
        zeroCopyLatency.record(latency, TimeUnit.NANOSECONDS);
        
        DistributionSummary.builder("gateway.request.size")
            .register(meterRegistry)
            .record(size);
    }
}

public class GatewayLogger {
    
    private static final Logger log = LoggerFactory.getLogger(GatewayLogger.class);
    
    public void logRequest(ServerWebExchange exchange, long duration) {
        ServerHttpRequest request = exchange.getRequest();
        
        // 只記錄元數(shù)據(jù)
        log.info("請求處理完成: path={}, method={}, duration={}ms, size={}",
            request.getPath().value(),
            request.getMethod(),
            duration,
            request.getHeaders().getContentLength());
    }
    
    public void logValidationResult(boolean isValid, String reason) {
        if (!isValid) {
            log.warn("參數(shù)驗(yàn)證失敗: {}", reason);
        }
    }
}

// 1. 開啟Netty內(nèi)存泄漏檢測
// 啟動參數(shù): -Dio.netty.leakDetection.level=PARANOID

// 2. 使用try-with-resources確保資源釋放
public void safeProcess(ByteBuf buffer) {
    try (ManagedResource resource = new ManagedResource(buffer)) {
        process(resource.getView());
    }  // 自動釋放
}

// 3. 定期監(jiān)控
@Scheduled(fixedRate = 60000)
public void monitorMemory() {
    BufferAllocatorMetric metric = PooledByteBufAllocator.DEFAULT.metric();
    long usedDirectMemory = metric.usedDirectMemory();
    
    if (usedDirectMemory > 100 * 1024 * 1024) { // 100MB閾值
        log.warn("直接內(nèi)存使用過高: {} bytes", usedDirectMemory);
    }
}

public class ZeroCopyErrorHandler {
    
    public Mono<Void> handleWithFallback(ServerWebExchange exchange, Throwable error) {
        if (error instanceof IllegalReferenceCountException) {
            // 引用計(jì)數(shù)異常，可能的內(nèi)存泄漏
            log.error("引用計(jì)數(shù)異常", error);
            return sendError(exchange, "系統(tǒng)異常");
        }
        
        if (error instanceof TimeoutException) {
            // 驗(yàn)證服務(wù)超時
            log.warn("驗(yàn)證服務(wù)超時");
            return sendError(exchange, "驗(yàn)證服務(wù)超時");
        }
        
        if (error instanceof DataBufferLimitException) {
            // 請求體過大
            log.warn("請求體過大: {}", error.getMessage());
            return sendError(exchange, "請求體過大");
        }
        
        // 其他異常，回退到傳統(tǒng)方案
        return fallbackToHeapCopy(exchange);
    }
    
    private Mono<Void> fallbackToHeapCopy(ServerWebExchange exchange) {
        // 回退到堆內(nèi)存拷貝方案
        log.warn("零拷貝失敗，回退到堆拷貝");
        return traditionalValidationFilter.filter(exchange, chain);
    }
}