RocketMQ4.5.X 实现修改生产者消费者日志保存路径

RocketMQ修改生产者消费者日志保存路径

rocket默认是将所有日志文件保存到user.home的对于win系统就是C盘了。

1.修改RocketMQ中CLientLogger.class的源码，把经过更改的源码重新打包后，去自己的maven仓库替换rocketmq-client.-4.5.X.jar。

2.对于生产者的启动类里需要配置JVM系统属性：

如果不设置logUserSlf4j为true的话，启动生产者的时候会报找不到日志配置文件的警告。

3.对于消费者，仅仅添加rq.lordir（日志保存路径）就行了。

当然，以上情况是开发中每个开发人员需要独立设置的，部署到线上的时候可以直接在源码中写路径而不是系统属性。

RocketMQ 日志操作

官网：http://rocketmq.apache.org/docs/logappender-example/

应用：将程序日志输出到rocketmq，消费端可读取日志数据进行相应处理

导入 jar 包

<dependency>
   <groupId>org.apache.rocketmq</groupId>
   <artifactId>rocketmq-logappender</artifactId>
   <version>4.7.1</version>
</dependency>

相关类

RocketmqLogbackAppender：发送到rocketmq

public class RocketmqLogbackAppender extends AppenderBase<ILoggingEvent> {
   private String tag;                   //标签
   private String topic;                 //发送的topic
   private String nameServerAddress;     //namesrv地址
   private String producerGroup;         //发送群组
   private MQProducer producer;
   private Layout layout;                //布局方式
   private PreSerializationTransformer<ILoggingEvent> pst = new LoggingEventPreSerializationTransformer();

AsyncAppender：异步发送

public class AsyncAppender extends AsyncAppenderBase<ILoggingEvent> {
   boolean includeCallerData = false;

AsyncAppenderBase

public class AsyncAppenderBase<E> extends UnsynchronizedAppenderBase<E>
implements AppenderAttachable<E> {
   AppenderAttachableImpl<E> aai = new AppenderAttachableImpl<E>();
   BlockingQueue<E> blockingQueue;
   public static final int DEFAULT_QUEUE_SIZE = 256;
   int queueSize = DEFAULT_QUEUE_SIZE;
   int appenderCount = 0;
   static final int UNDEFINED = -1;
   int discardingThreshold = UNDEFINED;
   boolean neverBlock = false;
   Worker worker = new Worker();
   public static final int DEFAULT_MAX_FLUSH_TIME = 1000;
   int maxFlushTime = DEFAULT_MAX_FLUSH_TIME;

AppenderAttachable：引用、删除appender

public interface AppenderAttachable<E> {
   void addAppender(Appender<E> newAppender);
   boolean isAttached(Appender<E> appender);

Iterator<Appender<E>> iteratorForAppenders();
   Appender<E> getAppender(String name);

void detachAndStopAllAppenders();
   boolean detachAppender(Appender<E> appender);
   boolean detachAppender(String name);
}

日志配置

logback.xml：放在resources目录下

<?xml version="1.0" encoding="UTF-8"?>
<configuration debug="false">
   <conversionRule conversionWord="clr" converterClass="org.springframework.boot.logging.logback.ColorConverter"/>
   <conversionRule conversionWord="wex" converterClass="org.springframework.boot.logging.logback.WhitespaceThrowableProxyConverter"/>
   <conversionRule conversionWord="wEx" converterClass="org.springframework.boot.logging.logback.ExtendedWhitespaceThrowableProxyConverter"/>

<property name="log.path" value="e:/hhhh/"/>
   <property name="console.pattern" value="％clr(％d{${LOG_DATEFORMAT_PATTERN:-yyyy-MM-dd HH:mm:ss.SSS}}){faint} ％clr(${LOG_LEVEL_PATTERN:-％5p}) ％clr(${PID:- }){magenta} ％clr(---){faint} ％clr([％15.15t]){faint} ％clr(％-40.40logger{39}){cyan} ％clr(:){faint} ％m％n${LOG_EXCEPTION_CONVERSION_WORD:-％wEx}"/>
   <property name="file.pattern" value="％d{${LOG_DATEFORMAT_PATTERN:-yyyy-MM-dd HH:mm:ss.SSS}} ${LOG_LEVEL_PATTERN:-％5p} ${PID:- } --- [％t] ％-40.40logger{39} : ％m％n${LOG_EXCEPTION_CONVERSION_WORD:-％wEx}"/>

<appender name="console" class="ch.qos.logback.core.ConsoleAppender">
       <encoder>
           <pattern>${console.pattern}</pattern>
       </encoder>
   </appender>

<appender name="mqAppender1" class="org.apache.rocketmq.logappender.logback.RocketmqLogbackAppender">
       <tag>tag</tag>
       <topic>topic-log</topic>
       <producerGroup>log-group</producerGroup>
       <nameServerAddress>172.18.0.10:9876</nameServerAddress>
       <layout>
           <pattern>％date ％p ％t - ％m％n</pattern>
       </layout>
   </appender>

<!-- 异步发送 -->
   <appender name="mqAsyncAppender1" class="ch.qos.logback.classic.AsyncAppender">
       <queueSize>1024</queueSize>
       <discardingThreshold>80</discardingThreshold>
       <maxFlushTime>2000</maxFlushTime>
       <neverBlock>true</neverBlock>
       <appender-ref ref="mqAppender1"/>
   </appender>

<root level="info">
       <appender-ref ref="console"/>
       <appender-ref ref="mqAsyncAppender1"/>
   </root>
</configuration>

将springboot应用打包，发布成docker容器

创建容器

#namesrv
docker run -it -d --net fixed --ip 172.18.0.10 -p 9876:9876 \
-e JAVA_OPT="-server -Xms256m -Xmx256m -Xmn128m" \
--name namesrv lihu12344/rocketmq:4.7.1 bash bin/mqnamesrv

#broker
docker run -it -d --net fixed --ip 172.18.0.20 \
-e NAMESRV_ADDR="172.18.0.10:9876" \
-e JAVA_OPT="-server -Xms512m -Xmx512m -Xmn256m" \
-v /usr/rocketmq/single/broker.conf:/home/rocketmq/rcketmq-4.7.1/conf/broker.conf \
--name broker lihu12344/rocketmq:4.7.1 \
bash bin/mqbroker autoCreateTopicEnable=true -c conf/broker.conf

#应用程序
docker run -it -d --net fixed --ip 172.18.0.21 -p 8080:8080 --name rocketmq-log rocketmq-log

rocketmq监控

docker run -it -d --net fixed --ip 172.18.0.4 -p 8008:8080 \
-e "JAVA_OPTS=-Drocketmq.namesrv.addr=172.18.0.10:9876 -Dcom.rocketmq.sendMessageWithVIPChannel=false" \
--name rocketmq-console styletang/rocketmq-console-ng

使用测试

192.168.57.127:8008

查看topic

查看topic-log message信息

来源：https://blog.csdn.net/qq_35251168/article/details/99695159