分布式日志治理：Log4j2自定义Appender写日志到RocketMQ

news2025/5/20 4:42:02

🧑 博主简介：CSDN博客专家，历代文学网（PC端可以访问：https://literature.sinhy.com/#/?__c=1000，移动端可微信小程序搜索“历代文学”）总架构师，15年工作经验，精通Java编程，高并发设计，Springboot和微服务，熟悉Linux，ESXI虚拟化以及云原生Docker和K8s，热衷于探索科技的边界，并将理论知识转化为实际应用。保持对新技术的好奇心，乐于分享所学，希望通过我的实践经历和见解，启发他人的创新思维。在这里，我希望能与志同道合的朋友交流探讨，共同进步，一起在技术的世界里不断学习成长。
技术合作请加本人wx（注明来自csdn）：foreast_sea

在这里插入图片描述

在这里插入图片描述

文章目录

  • Log4j2自定义Appender写日志到RocketMQ
      • 引言：分布式系统下的日志治理新范式——基于Log4j2与RocketMQ的高效实践
      • 1. 添加Maven依赖
      • 2. 实现自定义Appender
      • 3. 自定义Appender插件注册
      • 4. 配置log4j2.xml
      • 5. 关键点说明
      • 6. 注意事项

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;
    }

    @Override
    public 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();
    }

    @Override
    public 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);
        }
    }

    @Override
    public void stop() {
        super.stop();
        if (producer != null) {
            try {
                producer.close();
            } catch (Exception e) {
                LOGGER.error("Close RocketMQ Producer failed", e);
            }
        }
    }

    @PluginFactory
    public 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. 自定义Appender插件注册

log4j2最新版本的插件注册通过将Log4j插件描述文件（即Log4j2Plugins.dat）放入类路径完成。该文件在编译时通过PluginProcessor注解处理器生成。

需按以下方式配置构建工具以启用PluginProcessor：
Maven配置

<build>
    <plugins>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-compiler-plugin</artifactId>
            <executions>
                <execution>
                    <id>generate-log4j-plugin-descriptor</id>
                    <goals>
                        <goal>compile</goal>
                    </goals>
                    <phase>process-classes</phase>
                    <configuration>
                        <proc>only</proc>
                        <annotationProcessorPaths>
                        
                            <!-- 引入包含`PluginProcessor`的`log4j-core`，用于生成`Log4j2Plugins.dat` -->
                            <path>
                                <groupId>org.apache.logging.log4j</groupId>
                                <artifactId>log4j-core</artifactId>
                                <version>2.24.3</version>
                            </path>
                        </annotationProcessorPaths>
                        <annotationProcessors>
                            <!-- 使用`PluginProcessor`处理源码并生成`Log4j2Plugins.dat` -->
                            <processor>org.apache.logging.log4j.core.config.plugins.processor.PluginProcessor</processor>
                        </annotationProcessors>
                    </configuration>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

编译后，classes/META-INF目录下会多出一个org目录
在这里插入图片描述
该目录下存放的就是插件的注册信息：
在这里插入图片描述

4. 配置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>

5. 关键点说明

  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属性
    • 添加消息压缩功能
    • 支持同步/异步发送模式切换

6. 注意事项

  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等全链路追踪信息

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/2337380.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

【口腔粘膜鳞状细胞癌】文献阅读3

【口腔粘膜鳞状细胞癌】文献阅读3

文献 Single-cell transcriptomic analysis uncovers the origin and intratumoral heterogeneity of parotid pleomorphic adenoma 单细胞转录组学分析揭示了腮腺多形性腺瘤的起源和瘤内异质性 IF:10.8中科院分区:1区 医学WOS分区:Q1 摘要 多形性腺瘤 &#xff08;PA&#…
阅读更多...
RL中的rollout和episode的区别请问是啥

RL中的rollout和episode的区别请问是啥

很好的问题兄弟&#xff0c;rollout 和 episode 在强化学习&#xff08;RL&#xff09;里经常一起出现&#xff0c;虽然有重叠&#xff0c;但含义和使用语境还是有区别的&#xff1a; ✅ 一句话总结&#xff1a; Episode 是一个完整的任务过程&#xff08;从起点到终点&#xf…
阅读更多...
个人博客系统后端 - 用户信息管理功能实现指南（上）

个人博客系统后端 - 用户信息管理功能实现指南（上）

本文记录了如何实现用获取户信息&#xff0c;用户信息更新&#xff0c;用户头像上传三大基础功能 先上接口实现截图&#xff1a; 一、项目结构概览 先介绍一下 个人博客系统采用了标准的 Spring Boot 项目结构&#xff0c;用户功能相关的文件主要分布在以下几个目录&#xff1a…
阅读更多...
具身智能机器人学习路线全解析

具身智能机器人学习路线全解析

一、引言 具身智能机器人作为融合了机器人学、人工智能、认知科学等多领域知识的前沿技术&#xff0c;正逐渐改变着我们的生活和工作方式。从工业制造到家庭服务&#xff0c;从医疗护理到太空探索&#xff0c;具身智能机器人都展现出了巨大的潜力。对于想要深入了解和学习这一…
阅读更多...
写项目时一些疑惑：组件间的通信、createDownloadUrl和DownloadUrl，ArrayBuffer与Blob等

写项目时一些疑惑：组件间的通信、createDownloadUrl和DownloadUrl，ArrayBuffer与Blob等

目录 一、[vite] Internal server error: No known conditions for "./lib/locale/lang/zh-cn" specifier in "element-plus" package 二、可以用vue和JS的代码片段，但是用不了html的代码片段 三、meta是什么东西 四、为什么代码保持一致，但是时间轴始…
阅读更多...
手动关闭ArcGIS与ArcGIS Online连接的方法

手动关闭ArcGIS与ArcGIS Online连接的方法

【关闭软件启动时ArcGIS与ArcGIS Online连接方法】 打开C盘找到文件夹“C:\Program Files (x86)\Common Files\ArcGIS\bin”&#xff0c;如下图&#xff0c;删除“ArcGISConnection.exe”与“ArcGISConnectionTest.exe”文件&#xff0c;软件下次启动的时候就不会建立与ArcGIS …
阅读更多...
大M法处理非线性约束线性化

大M法处理非线性约束线性化

在电力系统优化问题中&#xff0c;大M法&#xff08;Big M Method&#xff09;是一种经典的处理非线性约束线性化的技术&#xff0c;尤其适用于混合整数线性规划&#xff08;MILP&#xff09;问题。 其核心思想是通过引入足够大的常数M和辅助变量&#xff08;如二元变量或松弛…
阅读更多...
【网络安全】谁入侵了我的调制解调器？（一）

【网络安全】谁入侵了我的调制解调器？（一）

文章目录 我被黑了159.65.76.209，你是谁？黑客攻击黑客？交出证据三年后我被黑了 两年前，在我家里使用家庭网络远程办公时，遇到了一件非常诡异的事情。当时，我正在利用一个“盲 XXE 漏洞”，这个漏洞需要借助一个外部 HTTP 服务器来“走私”文件。为了实现这一点，我在 AW…
阅读更多...
【Nokia 7360 ISAM局端】7360局端升级步骤

【Nokia 7360 ISAM局端】7360局端升级步骤

引言 Nokia 7360 ISAM局端是当前主流的OLT局端之一，在测试ONT产品中经常需要对接7360局端，特别是欧美等海外运营商。测试过程中经常需要升级OLT版本，以便对齐前方客户的现网环境。本文介绍将Nokia 7360 ISAM局端升级到L6GPAA65.669版本的详细步骤。 连接带外管理口 将维护…
阅读更多...
主数据管理：企业数字化转型的 “数据基石“ 如何为 AI 筑基？

主数据管理：企业数字化转型的 “数据基石“ 如何为 AI 筑基？

引言&#xff1a;当数据成为新石油&#xff0c;谁在炼制 "高纯度燃料"&#xff1f; 在数字化转型的浪潮中&#xff0c;企业宛如行驶在数据海洋中的巨轮&#xff0c;AI 则是驱动巨轮破浪前行的引擎。但引擎能否高效运转&#xff0c;取决于燃料的纯度 —— 这正是主数…
阅读更多...
使用 chromedriver 实现网络爬虫【手抄】

使用 chromedriver 实现网络爬虫【手抄】

1、引用 selenium 包 <dependency><groupId>org.seleniumhq.selenium</groupId><artifactId>selenium-java</artifactId><version>4.29.0</version> </dependency> <dependency><groupId>org.seleniumhq.seleniu…
阅读更多...
Linux之 grep、find、ls、wc 命令

Linux之 grep、find、ls、wc 命令

Linux之 grep、find、ls、wc 命令 “ 在 Linux 世界中&#xff0c;命令行是不可或缺的一部分&#xff0c;而掌握一些常用的命令可以帮助你更有效率地管理文件和系统。本文将为你介绍四個基礎而强大的 Linux 命令&#xff1a;grep、find、ls 和 wc&#xff0c;带你开启高效文件…
阅读更多...
AI 模型高效化：推理加速与训练优化的技术原理与理论解析

AI 模型高效化：推理加速与训练优化的技术原理与理论解析

AI 模型高效化&#xff1a;推理加速与训练优化的技术原理与理论解析 文章目录 AI 模型高效化&#xff1a;推理加速与训练优化的技术原理与理论解析一、推理加速&#xff1a;让模型跑得更快的“程序员魔法”&#xff08;一&#xff09;动态结构自适应推理&#xff1a;像人类一样…
阅读更多...
c++STL——vector的使用和模拟实现

c++STL——vector的使用和模拟实现

文章目录 vector的使用和模拟实现vector的使用vector介绍重点接口的讲解迭代器部分默认成员函数空间操作增删查改操作迭代器失效问题(重要)调整迭代器 vector的模拟实现实现的版本模拟实现结构预先处理的函数尾插函数push_backswap函数赋值重载size函数reserve函数 迭代器默认成…
阅读更多...
git更新的bug

git更新的bug

文章目录 1. 问题2. 分析 1. 问题 拉取了一个项目后遇到了这个问题&#xff0c; nvocation failed Server returned invalid Response. java.lang.RuntimeException: Invocation failed Server returned invalid Response. at git4idea.GitAppUtil.sendXmlRequest(GitAppUtil…
阅读更多...
github | 仓库权限管理 | 开权限

github | 仓库权限管理 | 开权限

省流版总结&#xff1a; github 给别人开权限&#xff1a;仓库 -> Setting -> Cllaborate -> Add people GitHub中 将公开仓库改为私有&#xff1a;仓库 -> Setting -> Danger Zone&#xff08;危险区&#xff09; ->Change repository visibility( 更改仓…
阅读更多...
uniapp自定义底部导航栏，解决下拉时候顶部空白的问题

uniapp自定义底部导航栏，解决下拉时候顶部空白的问题

一、背景 最近使用uniapp开发微信小程序&#xff0c;因为使用了自定义的顶部导航栏&#xff0c;所以在ios平台上&#xff08;Android未测试&#xff09;测试的时候&#xff0c;下拉的时候会出现整个页面下拉并且顶部留下大片空白的问题 二、任务&#xff1a;解决这个问题 经…
阅读更多...
C++学习之密码学知识

C++学习之密码学知识

目录 1.文档介绍 2.知识点概述 3.项目准备 4.序列化介绍 5.项目中基础组件介绍 6.基础模块在项目中作用 7.项目中其他模块介绍 8.加密三要素 9.对称加密和非堆成加密 10.对称和非对称加密特点 11.堆成加密算法des 12.des对称加密算法 13.对称加密算法aes 14.知识点…
阅读更多...
力扣 797. 所有可能的路径

力扣 797. 所有可能的路径

题目 给你一个有 n 个节点的 有向无环图&#xff08;DAG&#xff09;&#xff0c;请你找出所有从节点 0 到节点 n-1 的路径并输出&#xff08;不要求按特定顺序&#xff09; graph[i] 是一个从节点 i 可以访问的所有节点的列表&#xff08;即从节点 i 到节点 graph[i][j]存在一…
阅读更多...
第二篇：linux之Xshell使用及相关linux操作

第二篇：linux之Xshell使用及相关linux操作

第二篇&#xff1a;linux之Xshell使用及相关linux操作 文章目录 第二篇&#xff1a;linux之Xshell使用及相关linux操作一、Xshell使用1、Xshell安装2、Xshell使用 二、Bash Shell介绍与使用1、什么是Bash Shell(壳)&#xff1f;2、Bash Shell能干什么&#xff1f;3、平时如何使…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023