🧑 博主簡介：CSDN博客專家，歷代文學網（PC端可以訪問：https://literature.sinhy.com/#/?__c=1000，移動端可微信小程序搜索“歷代文學”）總架構師，15年工作經驗，精通Java編程，高并發設計，Springboot和微服務，熟悉Linux，ESXI虛擬化以及云原生Docker和K8s，熱衷于探索科技的邊界，并將理論知識轉化為實際應用。保持對新技術的好奇心，樂于分享所學，希望通過我的實踐經歷和見解，啟發他人的創新思維。在這里，我希望能與志同道合的朋友交流探討，共同進步，一起在技術的世界里不斷學習成長。
技術合作請加本人wx（注明來自csdn）：foreast_sea

---

Log4j2自定義Appender寫日志到RocketMQ

引言：分布式系統下的日志治理新范式——基于Log4j2與RocketMQ的高效實踐

在云原生與微服務架構大行其道的今天，日志管理已從簡單的本地文件存儲演化為支撐系統可觀測性的核心支柱。傳統日志處理方式在面對日均TB級的日志量、跨地域服務調用鏈追蹤、實時異常檢測等場景時，往往陷入存儲碎片化、檢索效率低下、處理延遲高的困境。尤其在金融交易、物聯網、在線教育等高并發領域，日志數據不僅是問題排查的"黑匣子"，更是業務洞察的"數據金礦"，亟需一種能夠兼顧實時性、可靠性和可擴展性的新型日志處理方案。

Apache RocketMQ作為阿里巴巴開源的高性能分布式消息中間件，憑借其毫秒級消息投遞、萬億級消息堆積能力和完善的事務機制，為日志數據的異步化處理提供了理想通道。而Log4j2作為Java生態中最主流的日志框架，其插件化架構和異步日志特性，使得開發者能夠通過自定義Appender將日志生產與傳輸邏輯解耦。二者的結合，不僅實現了日志從"被動記錄"到"主動流轉"的范式升級，更構建起日志采集、傳輸、存儲、分析的全鏈路解決方案。

本文深入探討如何基于Log4j2最新架構擴展日志輸出能力，通過構建自定義RocketMQAppender實現日志數據的實時投遞。該方案突破傳統日志文件的物理邊界，使日志數據可無縫對接Elasticsearch、Flink、Spark等大數據處理平臺，為實時監控、安全審計、用戶行為分析等場景提供高時效數據源。

本文從Maven依賴配置、Appender線程模型設計、RocketMQ生產者最佳實踐等維度展開，詳細解析如何在高并發場景下保障日志傳輸的可靠性與性能平衡，并針對消息壓縮、失敗重試、資源監控等關鍵問題給出工程級解決方案。通過此實踐，開發者可將日志系統的吞吐量提升1-2個數量級，同時顯著降低日志丟失風險，為構建企業級可觀測性平臺奠定堅實基礎。

以下是基于Java Log4j2自定義Appender將日志寫入RocketMQ的步驟：

1. 添加Maven依賴

<!-- Log4j2 核心依賴 -->
<dependency><groupId>org.apache.logging.log4j</groupId><artifactId>log4j-core</artifactId><version>2.23.1</version>
</dependency><!-- RocketMQ客戶端 -->
<dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-client</artifactId><version>5.1.4</version>
</dependency>

2. 實現自定義Appender

import org.apache.logging.log4j.core.*;
import org.apache.logging.log4j.core.appender.AbstractAppender;
import org.apache.logging.log4j.core.config.Property;
import org.apache.logging.log4j.core.config.plugins.*;
import org.apache.rocketmq.client.apis.*;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.apache.rocketmq.client.apis.producer.ProducerBuilder;
import org.apache.rocketmq.client.apis.producer.SendResult;import java.io.Serializable;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;@Plugin(name = "RocketMQAppender",category = Core.CATEGORY_NAME,elementType = Appender.ELEMENT_TYPE,printObject = true
)
public final class RocketMQAppender extends AbstractAppender {private Producer producer;private final String namesrvAddr;private final String topic;private final String producerGroup;private final int sendTimeout;protected RocketMQAppender(String name, Filter filter, Layout<? extends Serializable> layout,String namesrvAddr, String topic, String producerGroup, int sendTimeout) {super(name, filter, layout, true, Property.EMPTY_ARRAY);this.namesrvAddr = namesrvAddr;this.topic = topic;this.producerGroup = producerGroup;this.sendTimeout = sendTimeout;}@Overridepublic void start() {try {final ClientServiceProvider provider = ClientServiceProvider.loadService();ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(namesrvAddr);ProducerBuilder producerBuilder = provider.newProducerBuilder().setClientConfiguration(builder.build()).setTopics(topic);if (producerGroup != null) {producerBuilder.setProducerGroup(producerGroup);}producer = producerBuilder.build();} catch (ClientException e) {LOGGER.error("Initialize RocketMQ Producer failed", e);}super.start();}@Overridepublic void append(LogEvent event) {if (producer == null) return;try {byte[] body = getLayout().toByteArray(event);String messageBody = new String(body, StandardCharsets.UTF_8);final ClientServiceProvider provider = ClientServiceProvider.loadService();Message message = provider.newMessageBuilder().setTopic(topic).setBody(body).build();SendResult sendResult = producer.send(message);// 可添加發送結果處理邏輯} catch (Exception e) {LOGGER.error("Send log to RocketMQ failed", e);}}@Overridepublic void stop() {super.stop();if (producer != null) {try {producer.close();} catch (Exception e) {LOGGER.error("Close RocketMQ Producer failed", e);}}}@PluginFactorypublic static RocketMQAppender createAppender(@PluginAttribute("name") String name,@PluginElement("Filter") Filter filter,@PluginElement("Layout") Layout<? extends Serializable> layout,@PluginAttribute("namesrvAddr") String namesrvAddr,@PluginAttribute("topic") String topic,@PluginAttribute(value = "producerGroup", defaultString = "LogProducerGroup") String producerGroup,@PluginAttribute(value = "sendTimeout", defaultInt = 3000) int sendTimeout) {if (name == null) {LOGGER.error("No name provided for RocketMQAppender");return null;}return new RocketMQAppender(name, filter, layout, namesrvAddr, topic, producerGroup, sendTimeout);}
}

3. 配置log4j2.xml

<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="WARN"><Appenders><RocketMQAppender name="RocketMQ"namesrvAddr="localhost:8081"topic="LOG_TOPIC"producerGroup="LOG_PRODUCER_GROUP"sendTimeout="5000"><PatternLayout pattern="%d{yyyy-MM-dd HH:mm:ss} %-5level %logger{36} - %msg%n"/></RocketMQAppender></Appenders><Loggers><Root level="info"><AppenderRef ref="RocketMQ"/></Root></Loggers>
</Configuration>

4. 關鍵點說明

1. 線程安全設計：

   • RocketMQ Producer是線程安全的，可以復用實例
   • 在start()中初始化，stop()中銷毀

2. 異常處理：

   • 在send方法中添加try-catch防止日志記錄阻塞主線程
   • 建議添加失敗重試機制（示例未展示）

3. 性能優化建議：

   // 可添加批量發送支持
   producer.send(List<Message> messages, SendReceipt sendReceipt);// 或使用異步發送
   CompletableFuture<SendResult> future = producer.sendAsync(message);
   

4. 擴展功能建議：

   • 添加消息Tag支持
   • 支持自定義Key/Value屬性
   • 添加消息壓縮功能
   • 支持同步/異步發送模式切換

5. 注意事項

1. 版本兼容性：

   • RocketMQ 5.x+ 使用新的客戶端API
   • 舊版本（4.x）需要調整客戶端實現

2. 資源管理：

   • 確保Producer在JVM關閉時正確關閉
   • 建議添加發送隊列積壓監控

3. 安全配置：

   // 如果需要認證
   ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(namesrvAddr).setCredentialProvider(new StaticSessionTokenCredentialProvider("accessKey", "secretKey"));
   

4. 日志格式化：

   • 建議使用JSON格式方便后續處理
   • 可添加TraceID等全鏈路追蹤信息