Rocketmq日志收集与logback集成Demo

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

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

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

• 这是生产端打印
  

• 这是消费端打印
  

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