Rocketmq日志收集与logback集成Demo

时间：2022-01-20 22:32:59

官方文档有简洁的例子，这里就做一个简单补充和实践

  1. 直接上logback-boot.xml文件
<?xml version="1.0" encoding="UTF-8"?>

<configuration>

    <!-- 文件输出格式 -->

    <property name="PATTERN" value="%-12(%d{yyyy-MM-dd HH:mm:ss.SSS}) |-%-5level [%thread] %c [%L] -| %msg%n"/>

    <!-- 生产日志文件路径 -->

    <property name="LOG_FILE_PATH" value="/home/tomapp/manager/logs"/>

    <!-- 开发环境 -->

    <springProfile name="dev">

        <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">

            <encoder>

                <pattern>${PATTERN}</pattern>

            </encoder>

        </appender>

        <logger name="com.tzxylao.manager" level="debug"/>

        <root level="info">

            <appender-ref ref="CONSOLE"/>

        </root>

    </springProfile>

    <!-- 生产环境 -->

    <springProfile name="prod">

        <!-- 每天产生一个文件 -->

        <appender name="PROD_FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">

            <!-- 文件路径 -->

            <file>${LOG_FILE_PATH}/manager.log</file>

            <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">

                <!-- 文件名称 -->

                <fileNamePattern>${LOG_FILE_PATH}/manager.%d{yyyy-MM-dd}.log</fileNamePattern>

                <!-- 文件最大保存历史数量 -->

                <MaxHistory>100</MaxHistory>

            </rollingPolicy>

            <layout class="ch.qos.logback.classic.PatternLayout">

                <pattern>${PATTERN}</pattern>

            </layout>

        </appender>

        <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">

            <encoder>

                <pattern>${PATTERN}</pattern>

            </encoder>

        </appender>

        <appender name="mqAppender1" class="org.apache.rocketmq.logappender.logback.RocketmqLogbackAppender">

            <tag>TagA</tag>

            <topic>TopicTest</topic>

            <producerGroup>please_rename_unique_group_name</producerGroup>

            <nameServerAddress>47.101.45.25: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>

        <logger name="com.tzxylao.manager" level="debug"/>

        <root level="info">

            <appender-ref ref="PROD_FILE"/>

            <appender-ref ref="CONSOLE"/>

            <appender-ref ref="mqAppender1"/>

        </root>

    </springProfile>

</configuration>
  • 这里有我本来的日志输出，rocketmq的输出主要是appender为mqAppender1、mqAsyncAppender1的这两个，最后放入root节点就可以了
  • 官方文档路径：https://rocketmq.apache.org/docs/logappender-example/
  • 这里要起作用，必须把nameServerAddress，rocketmq服务地址配上
  • 可能出错的地方就是找不到对应的类了，这里给出我的引用pom
<dependency>

    <groupId>org.apache.rocketmq</groupId>

    <artifactId>rocketmq-client</artifactId>

    <version>4.3.0</version>

    <exclusions>

        <exclusion>

            <groupId>org.slf4j</groupId>

            <artifactId>slf4j-api</artifactId>

        </exclusion>

    </exclusions>

</dependency>

<dependency>

    <groupId>org.apache.rocketmq</groupId>

    <artifactId>rocketmq-logappender</artifactId>

    <version>4.3.0</version>

</dependency>

测试

  1. 编写个简单的消费端服务，就拿官方的例子加上服务地址就行了
/*

 * Licensed to the Apache Software Foundation (ASF) under one or more

 * contributor license agreements.  See the NOTICE file distributed with

 * this work for additional information regarding copyright ownership.

 * The ASF licenses this file to You under the Apache License, Version 2.0

 * (the "License"); you may not use this file except in compliance with

 * the License.  You may obtain a copy of the License at

 *

 *     http://www.apache.org/licenses/LICENSE-2.0

 *

 * Unless required by applicable law or agreed to in writing, software

 * distributed under the License is distributed on an "AS IS" BASIS,

 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

 * See the License for the specific language governing permissions and

 * limitations under the License.

 */

package com.tzxylao.quick;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;

import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;

import org.apache.rocketmq.client.exception.MQClientException;

import org.apache.rocketmq.common.consumer.ConsumeFromWhere;

import org.apache.rocketmq.common.message.MessageExt;

import java.io.UnsupportedEncodingException;

import java.util.List;

/**

 * This example shows how to subscribe and consume messages using providing {@link DefaultMQPushConsumer}.

 */

public class Consumer {

    public static void main(String[] args) throws InterruptedException, MQClientException {

        /*

         * Instantiate with specified consumer group name.

         */

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4");

        /*

         * Specify name server addresses.

         * <p/>

         *

         * Alternatively, you may specify name server addresses via exporting environmental variable: NAMESRV_ADDR

         * <pre>

         * {@code

         * consumer.setNamesrvAddr("name-server1-ip:9876;name-server2-ip:9876");

         * }

         * </pre>

         */

        /*

         * Specify where to start in case the specified consumer group is a brand new one.

         */

        consumer.setNamesrvAddr("47.101.45.25:9876");

        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        /*

         * Subscribe one more more topics to consume.

         */

        consumer.subscribe("TopicTest", "*");

        /*

         *  Register callback to execute on arrival of messages fetched from brokers.

         */

        consumer.registerMessageListener(new MessageListenerConcurrently() {

            @Override

            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,

                                                            ConsumeConcurrentlyContext context) {

//                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);

                try {

                    String s = new String(msgs.get(0).getBody(), "utf-8");

                    System.out.println(s);

                } catch (UnsupportedEncodingException e) {

                    e.printStackTrace();

                }

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

            }

        });

        /*

         *  Launch the consumer instance.

         */

        consumer.start();

        System.out.printf("Consumer Started.%n");

    }

}
  1. 好了，现在启动消费端，再启动配置了rocketmq日志的服务端

  • 这是生产端打印

    Rocketmq日志收集与logback集成Demo

  • 这是消费端打印

    Rocketmq日志收集与logback集成Demo

好了，日志收集后怎么用，自己发挥想象吧~~

相关文章