如何优化向Event Hub发送压缩Payload时的CPU利用率？

针对你遇到的4MB JSON payload压缩导致CPU利用率过高的问题，以下是几个可落地的优化方案：

1. 降低Gzip压缩级别 #

Gzip默认使用的压缩级别是Deflater.DEFAULT_COMPRESSION（级别6），这个级别在压缩率和速度之间取平衡，但对于CPU敏感场景，可切换到速度优先的低级别（比如级别1，Deflater.BEST_SPEED），能大幅降低CPU开销，虽然压缩率会略有下降，但仍能显著减少传输体积。

修改代码替换原GzipCompressingEntity的实现：

private void writeAndSendMetrics(StringBuilder resultJSON) throws URISyntaxException, IOException {
    String postBody = resultJSON.toString();
    if (!isJSONValid(postBody)) {
        logger.info("postBody is not a valid JSON {} ", postBody);
        return;
    }

    logger.debug("Request received for writeAndSendMetrics");
    HttpPost httpPost = new HttpPost(new URI(eventHubUrl));
    logger.debug("Result JSON: {}", resultJSON);

    long startTime = System.currentTimeMillis();

    StringEntity uncompressed = new StringEntity(postBody);
    logger.info("Uncompressed-size = " + uncompressed.getContentLength() );

    // 自定义Gzip压缩，指定速度优先级别
    ByteArrayOutputStream baos = new ByteArrayOutputStream();
    try (GZIPOutputStream gzipOut = new GZIPOutputStream(baos)) {
        // 设置压缩级别为BEST_SPEED（1），可选范围0-9，0为无压缩，9为最高压缩
        gzipOut.setLevel(Deflater.BEST_SPEED);
        uncompressed.writeTo(gzipOut);
    }
    ByteArrayEntity compressedEntity = new ByteArrayEntity(baos.toByteArray());
    compressedEntity.setContentEncoding("gzip");
    compressedEntity.setContentType(uncompressed.getContentType());
    httpPost.setEntity(compressedEntity);

    long endTime = System.currentTimeMillis();
    logger.info("Compression-Time = " + (endTime - startTime) + " ms" );

    httpPost.setHeader("Authorization", "xyz");
    org.apache.http.HttpResponse response = httpclient.execute(httpPost);
    EntityUtils.consumeQuietly(response.getEntity());
    logger.info("Response Status Code : {}", response.getStatusLine().getStatusCode());
}

2. 异步压缩解耦发送流程 #

将压缩操作从发送线程中剥离，放到独立的线程池执行，避免压缩占用发送线程的CPU资源，让发送线程专注于网络IO操作，同时平衡整体CPU负载。

示例代码：

// 初始化单线程压缩池（可根据CPU核心数调整，比如Runtime.getRuntime().availableProcessors()）
private final ExecutorService compressionExecutor = Executors.newSingleThreadExecutor(r -> {
    Thread t = new Thread(r);
    t.setDaemon(true);
    return t;
});

private void writeAndSendMetrics(StringBuilder resultJSON) throws URISyntaxException, IOException {
    String postBody = resultJSON.toString();
    if (!isJSONValid(postBody)) {
        logger.info("postBody is not a valid JSON {} ", postBody);
        return;
    }

    logger.debug("Request received for writeAndSendMetrics");
    StringEntity uncompressed = new StringEntity(postBody);
    logger.info("Uncompressed-size = " + uncompressed.getContentLength() );

    // 异步提交压缩+发送任务
    compressionExecutor.submit(() -> {
        long startTime = System.currentTimeMillis();
        try {
            ByteArrayOutputStream baos = new ByteArrayOutputStream();
            try (GZIPOutputStream gzipOut = new GZIPOutputStream(baos)) {
                gzipOut.setLevel(Deflater.BEST_SPEED);
                uncompressed.writeTo(gzipOut);
            }
            ByteArrayEntity compressedEntity = new ByteArrayEntity(baos.toByteArray());
            compressedEntity.setContentEncoding("gzip");
            compressedEntity.setContentType(uncompressed.getContentType());

            long endTime = System.currentTimeMillis();
            logger.info("Compression-Time = " + (endTime - startTime) + " ms" );

            // 执行发送
            HttpPost httpPost = new HttpPost(new URI(eventHubUrl));
            httpPost.setEntity(compressedEntity);
            httpPost.setHeader("Authorization", "xyz");
            org.apache.http.HttpResponse response = httpclient.execute(httpPost);
            EntityUtils.consumeQuietly(response.getEntity());
            logger.info("Response Status Code : {}", response.getStatusLine().getStatusCode());
        } catch (Exception e) {
            logger.error("Compression or send failed", e);
        }
    });
}

注意：确保httpclient是线程安全的（Apache的CloseableHttpClient本身支持多线程复用）。

3. 复用压缩资源减少对象开销 #

每次创建Deflater和GZIPOutputStream会带来对象初始化的CPU开销，通过ThreadLocal为每个线程复用一个Deflater实例，减少重复创建的成本。

示例代码：

// 线程本地复用Deflater，指定速度优先级别
private final ThreadLocal<Deflater> deflaterThreadLocal = ThreadLocal.withInitial(() -> {
    Deflater def = new Deflater(Deflater.BEST_SPEED);
    return def;
});

private void writeAndSendMetrics(StringBuilder resultJSON) throws URISyntaxException, IOException {
    // ... 前面的校验和初始化代码不变 ...

    StringEntity uncompressed = new StringEntity(postBody);
    logger.info("Uncompressed-size = " + uncompressed.getContentLength() );

    long startTime = System.currentTimeMillis();
    ByteArrayOutputStream baos = new ByteArrayOutputStream();
    Deflater def = deflaterThreadLocal.get();
    try (GZIPOutputStream gzipOut = new GZIPOutputStream(baos, def)) {
        uncompressed.writeTo(gzipOut);
    } finally {
        def.reset(); // 重置Deflater，以便下次复用
    }

    ByteArrayEntity compressedEntity = new ByteArrayEntity(baos.toByteArray());
    compressedEntity.setContentEncoding("gzip");
    compressedEntity.setContentType(uncompressed.getContentType());
    httpPost.setEntity(compressedEntity);

    long endTime = System.currentTimeMillis();
    logger.info("Compression-Time = " + (endTime - startTime) + " ms" );

    // ... 发送代码不变 ...
}

4. 替换为CPU友好的压缩算法 #

如果Event Hub支持非Gzip的压缩编码（比如Snappy、LZ4），可以替换为这些高速度、低CPU占用的算法，它们的压缩速度比Gzip快数倍，CPU开销大幅降低，虽然压缩率略低于Gzip，但对于JSON文本仍有不错的压缩效果。

比如使用Snappy压缩的示例（需引入Snappy库）：

// 引入Snappy依赖后，替换压缩逻辑
ByteArrayOutputStream baos = new ByteArrayOutputStream();
byte[] rawBytes = postBody.getBytes(StandardCharsets.UTF_8);
byte[] compressedBytes = Snappy.compress(rawBytes);
ByteArrayEntity compressedEntity = new ByteArrayEntity(compressedBytes);
compressedEntity.setContentEncoding("snappy");
compressedEntity.setContentType("application/json");
httpPost.setEntity(compressedEntity);

注意：需确认Event Hub是否支持snappy编码，若不支持则无法使用此方案。

内容的提问来源于stack exchange，提问作者Ankur Aggarwal