2. 程式人生 > >Kafka之——擴充套件logback將日誌輸出到Kafka例項

Kafka之——擴充套件logback將日誌輸出到Kafka例項

阿新 • • 發佈：2019-01-12

一、Kafka的安裝

注意：本文是基於kafka_2.11-0.9.0.0.tgz、zookeeper-3.4.9.tar.gz 實現的。

二、實現

1、建立Formatter介面

主要作用是定義一個format方法，接收ILoggingEvent物件，返回字串。具體程式碼如下：

package com.lyz.storm.formatter;

import ch.qos.logback.classic.spi.ILoggingEvent;

/**
 * 定義格式化介面
 * @author liuyazhuang
 *
 */
public interface Formatter {
	
	String format(ILoggingEvent event);
}

2、建立MessageFormatter類

主要是實現Formatter介面，主要程式碼如下：

package com.lyz.storm.formatter;

import ch.qos.logback.classic.spi.ILoggingEvent;

/**
 * 實現Formatter介面，接收ILoggingEvent物件返回一個字串供消費者處理
 * @author liuyazhuang
 *
 */
public class MessageFormatter implements Formatter {

	@Override
	public String format(ILoggingEvent event) {
		return event.getFormattedMessage();
	}

}

3、自定義KafkaAppender類

主要的作用就是講日誌重定向輸出到Kafka佇列中，主要程式碼如下：

package com.lyz.storm.appender;

import java.util.Properties;

import com.lyz.storm.formatter.Formatter;
import com.lyz.storm.formatter.MessageFormatter;

import ch.qos.logback.classic.spi.ILoggingEvent;
import ch.qos.logback.core.AppenderBase;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

/**
 * 自定義KafkaAppender將日誌輸出到Kafka中
 * @author liuyazhuang
 *
 */
public class KafkaAppender extends AppenderBase<ILoggingEvent> {
	
	private String topic;
	private String zookeeperHost;
	private Producer<String, String> producer;
	private Formatter formatter;
	private String brokerList;
	
	public String getTopic() {
		return topic;
	}

	public void setTopic(String topic) {
		this.topic = topic;
	}

	public String getZookeeperHost() {
		return zookeeperHost;
	}

	public void setZookeeperHost(String zookeeperHost) {
		this.zookeeperHost = zookeeperHost;
	}

	public Formatter getFormatter() {
		return formatter;
	}

	public void setFormatter(Formatter formatter) {
		this.formatter = formatter;
	}

	public String getBrokerList() {
		return brokerList;
	}

	public void setBrokerList(String brokerList) {
		this.brokerList = brokerList;
	}

	@Override
	public void start() {
		if(this.formatter == null){
			this.formatter = new MessageFormatter();
		}
		super.start();
		Properties props = new Properties();
		props.put("zk.connect", this.zookeeperHost);
	    props.put("serializer.class", "kafka.serializer.StringEncoder");
	    props.put("metadata.broker.list", this.brokerList);
	    ProducerConfig config = new ProducerConfig(props);
	    this.producer = new Producer<String, String>(config);
	}
	
	@Override
	public void stop() {
		super.stop();
		this.producer.close();
	}
	@Override
	protected void append(ILoggingEvent event) {
		String payload = this.formatter.format(event);
		KeyedMessage<String, String> data = new KeyedMessage<String, String>(this.topic, payload);
		this.producer.send(data);
	}

}

4、建立RogueApplication類

主要的作用就是模擬列印日誌，主要程式碼如下：

package com.lyz.storm.trend;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 模擬列印日誌
 * @author liuyazhuang
 *
 */
public class RogueApplication {
    private static final Logger LOG = LoggerFactory.getLogger("com.lyz.storm.trend.RogueApplication");
    public static void main(String[] args) throws Exception {
        int slowCount = 6;
        int fastCount = 15;
        while (true)        {
            for(int i = 0; i < slowCount; i++){
                LOG.warn("This is a warning (slow state).");
                Thread.sleep(5000);
            }
            for(int i = 0; i < fastCount; i++){
                LOG.warn("This is a warning (rapid state).");
                Thread.sleep(1000);
            }
            for(int i = 0; i < slowCount; i++){
                LOG.warn("This is a warning (slow state).");
                Thread.sleep(5000);
            }
        }

    }
}

5、建立logback.xml檔案

我們在專案的classpath目錄下建立logback.xml檔案，內容如下：

<?xml version="1.0" encoding="UTF-8"?>
<configuration>
   <appender name="KAFKA" class="com.lyz.storm.appender.KafkaAppender">
   		<topic>test</topic>
   		<zookeeperHost>192.168.209.121:2181</zookeeperHost>
   		<brokerList>192.168.209.121:9092</brokerList>
   </appender>
   <root level="debug">
   		<appender-ref ref="KAFKA"/>
   </root>
</configuration>

6、pom.xml

最後附上我們的pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.lyz</groupId>
    <artifactId>storm-chapter04</artifactId>
    <version>1.0</version>
    <name>storm-chapter04</name>


    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>
    <repositories>
        <repository>
            <id>clojars.org</id>
            <url>http://clojars.org/repo</url>
        </repository>
    </repositories>
    <dependencies>

        <!-- test -->
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.8.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.storm</groupId>
            <artifactId>storm-core</artifactId>
            <version>0.9.2-incubating</version>
            <exclusions>
                <exclusion>
                    <groupId>org.slf4j</groupId>
                    <artifactId>slf4j-log4j12</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>log4j</groupId>
                    <artifactId>log4j</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
     <dependency>
		    <groupId>org.apache.kafka</groupId>
		    <artifactId>kafka_2.11</artifactId>
		    <version>0.9.0.0</version>
		       <exclusions>  
                <exclusion>  
                    <groupId>org.apache.zookeeper</groupId>  
                    <artifactId>zookeeper</artifactId>  
                </exclusion>  
                <exclusion>  
                    <groupId>log4j</groupId>  
                    <artifactId>log4j</artifactId>  
                </exclusion>  
                <exclusion>  
                    <groupId>org.slf4j</groupId>  
                    <artifactId>slf4j-log4j12</artifactId>  
                </exclusion>  
            </exclusions> 
		</dependency>
        <dependency>
            <groupId>storm</groupId>
            <artifactId>storm-kafka</artifactId>
            <version>0.9.0-wip16a-scala292</version>
        </dependency>
        <dependency>
            <groupId>xmlpull</groupId>
            <artifactId>xmlpull</artifactId>
            <version>1.1.3.4a</version>
        </dependency>

        <dependency>
            <groupId>org.igniterealtime.smack</groupId>
            <artifactId>smack</artifactId>
            <version>3.2.1</version>
        </dependency>


        <dependency>
            <groupId>org.igniterealtime.smack</groupId>
            <artifactId>smackx</artifactId>
            <version>3.2.1</version>
        </dependency>
        
        <dependency>
		    <groupId>org.apache.hadoop</groupId>
		    <artifactId>hadoop-hdfs</artifactId>
		    <version>2.9.0</version>
		     <exclusions>
                <exclusion>
                    <groupId>org.slf4j</groupId>
                    <artifactId>slf4j-log4j12</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>log4j</groupId>
                    <artifactId>log4j</artifactId>
                </exclusion>
            </exclusions>
		</dependency>
   		<dependency>
		    <groupId>org.apache.hadoop</groupId>
		    <artifactId>hadoop-client</artifactId>
		    <version>2.9.0</version>
		     <exclusions>
                <exclusion>
                    <groupId>org.slf4j</groupId>
                    <artifactId>slf4j-log4j12</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>log4j</groupId>
                    <artifactId>log4j</artifactId>
                </exclusion>
            </exclusions>
		</dependency>
        
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-annotations</artifactId>
            <version>2.2.2</version>
        </dependency>
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
            <version>2.2.2</version>
        </dependency>
        
		<dependency>
			<groupId>com.metamx</groupId>
			<artifactId>java-util</artifactId>
			<version>0.26.12</version>
		</dependency>
    </dependencies>

   <build>
		<finalName>storm-chapter04</finalName>
		<resources>
			<resource>
				<targetPath>${project.build.directory}/classes</targetPath>
				<directory>src/main/resources</directory>
				<filtering>true</filtering>
				<includes>
					<include>**/*.xml</include>
					<include>**/*.properties</include>
				</includes>
			</resource>
			<!-- 結合com.alibaba.dubbo.container.Main -->
			<resource>
				<targetPath>${project.build.directory}/classes/META-INF/spring</targetPath>
				<directory>src/main/resources/spring</directory>
				<filtering>true</filtering>
				<includes>
					<include>spring-context.xml</include>
				</includes>
			</resource>
		</resources>
		
		<pluginManagement>
			<plugins>
				<!-- 解決Maven外掛在Eclipse內執行了一系列的生命週期引起衝突 -->
				<plugin>
					<groupId>org.eclipse.m2e</groupId>
					<artifactId>lifecycle-mapping</artifactId>
					<version>1.0.0</version>
					<configuration>
						<lifecycleMappingMetadata>
							<pluginExecutions>
								<pluginExecution>
									<pluginExecutionFilter>
										<groupId>org.apache.maven.plugins</groupId>
										<artifactId>maven-dependency-plugin</artifactId>
										  <versionRange>[2.0,)</versionRange>  
										<goals>
											<goal>copy-dependencies</goal>
										</goals>
									</pluginExecutionFilter>
									<action>
										<ignore />
									</action>
								</pluginExecution>
							</pluginExecutions>
						</lifecycleMappingMetadata>
					</configuration>
				</plugin>
			</plugins>
		</pluginManagement>
		<plugins>
			<!-- 打包jar檔案時，配置manifest檔案，加入lib包的jar依賴 -->
			<plugin>
				<groupId>org.apache.maven.plugins</groupId>
				<artifactId>maven-jar-plugin</artifactId>
				<configuration>
					<classesDirectory>target/classes/</classesDirectory>
					<archive>
						<manifest>
							<mainClass>com.lyz.chapter1.main.WordCountTopology</mainClass>
							<!-- 打包時 MANIFEST.MF檔案不記錄的時間戳版本 -->
							<useUniqueVersions>false</useUniqueVersions>
							<addClasspath>true</addClasspath>
							<classpathPrefix>lib/</classpathPrefix>
						</manifest>
						<manifestEntries>
							<Class-Path>.</Class-Path>
						</manifestEntries>
					</archive>
				</configuration>
			</plugin>
			<plugin>
				<groupId>org.apache.maven.plugins</groupId>
				<artifactId>maven-dependency-plugin</artifactId>
				<executions>
					<execution>
						<id>copy-dependencies</id>
						<phase>package</phase>
						<goals>
							<goal>copy-dependencies</goal>
						</goals>
						<configuration>
							<type>jar</type>
							<includeTypes>jar</includeTypes>
							<useUniqueVersions>false</useUniqueVersions>
							<outputDirectory>
								${project.build.directory}/lib
							</outputDirectory>
						</configuration>
					</execution>
				</executions>
			</plugin>
			
			 <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.1</version>
                <configuration>
                    <source>1.7</source>
                    <target>1.7</target>
                </configuration>
            </plugin>
			
		</plugins>

</build>
</project>

三、測試

在Kafka所在伺服器命令列輸入：

 kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning
監聽日誌輸出。

此時我們執行RogueApplication類，可以看到打印出如下日誌：

[[email protected] local]# kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning  
test
abcmytest
abcmytest
abc你好
abc你好
This is a warning (slow state).
This is a warning (slow state).
This is a warning (slow state).
This is a warning (slow state).
This is a warning (slow state).
This is a warning (slow state).
This is a warning (rapid state).
This is a warning (rapid state).
This is a warning (rapid state).
This is a warning (rapid state).
This is a warning (rapid state).
This is a warning (rapid state).
This is a warning (slow state).
This is a warning (slow state).
This is a warning (slow state).
This is a warning (slow state).
如下圖：


四、溫馨提示

相關推薦

Kafka之——擴充套件logback將日誌輸出到Kafka例項

一、Kafka的安裝 注意：本文是基於kafka_2.11-0.9.0.0.tgz、zookeeper-3.4.9.tar.gz 實現的。 二、實現 1、建立Formatter介面 主要作用是定義一個format方法，接收ILoggingEvent物件，返回字串。具體程

springboot2+logback將日誌輸出到oracle資料庫的踩坑之旅

背景 根據本人寫部落格的慣例，先交代下背景。在公司的系統中，我們的配置檔案是切分有好幾個的，不同的配置檔案裡面配置內容有著不同，對

將日誌輸出到屏幕及文件

bug ger round utf-8 col logger asc bin war #!/usr/bin/env python# -*- coding: utf-8 -*-import logging#創建日誌對像logger=logging.getLogger("TES

SLF4J + logback 實現日誌輸出和記錄

-- .com 保持 不存在 default stat 我們 fix jar包 一、SLF4J 　　SLF4J，即簡單日誌門面(Simple Logging Facade for Java)，不是具體的日誌解決方案，它只服務於各種各樣的日誌系統。在使用SLF4J的時候，不

Asp.Net Core NLog 將日誌輸出到數據庫以及添加LayoutRenderer的支持

tab 發現 exce images tex system 規則 .net pan 在這之前打算用Apache的Log4Net，但是發現其AdoNetAppender方法已經不存在了，無法使用配置文件直接輸出到數據庫了，因此我便改用了N

android 將日誌輸出到串口

ice dev 串口 service androi sys class group 代碼 在init.rc中添加如下代碼即可： service logcat /system/bin/logcat -f /dev/kmsg *:D 　　 class main

8、CentOS7 安裝Docker之擴充套件（將docker程序移動到一個不同的分割槽）注：操作雖然很簡單，但很實用。

將docker程序移動到一個不同的分割槽 Docker把與你的容器和映象相關的資料儲存到一個目錄下。由於可能會儲存潛在的大量不同的映象，這個目錄會很快變得很大！ 如果你的主機有不同的分割槽，你可能會更快遇到空間受限的問題。在這種情況下，你可能需要把Docker的資料目錄移動到其它分割槽。 問

logback將日誌寫入不同資料夾裡

轉載：logback不同業務的日誌列印到不同檔案 一、logback.xml檔案配置如下： <?xml version="1.0" encoding="UTF-8" ?> <configuration> <contextName>nana</cont

.NET Core的日誌[2]:將日誌輸出到控制檯

對於一個控制檯應用，比如採用控制檯應用作為宿主的ASP.NET Core應用，我們可以將記錄的日誌直接輸出到控制檯上。針對控制檯的Logger是一個型別為ConsoleLogger的物件，ConsoleLogger對應的LoggerProvider型別為ConsoleLoggerProvider，這兩個型別都

web專案Log4j日誌輸出路徑配置問題 問題描述：一個web專案想在一個tomcat下執行多個例項（通過修改war包名稱的實現），然後每個例項都將日誌輸出到tomcat的logs目錄下例項名命名的文

問題描述：一個web專案想在一個tomcat下執行多個例項（通過修改war包名稱的實現），然後每個例項都將日誌輸出到tomcat的logs目錄下例項名命名的資料夾下進行區分檢視每個例項日誌，要求通過儘可能少的改動配置檔案，最好修改例項名後可以不修改log4j的配置檔案。 實現分析：一般實現上面需求，需要在修

logback配置日誌輸出

<dependency> <groupId>log4j</groupId> <artifactId>log4j</artifactId> <version>1.2.17</ve

windows下使用log4j將日誌輸出到檔案，包括system.out

1、將debug級別和error級別分別輸出到檔案 log4j.rootLogger=info,console,file1,file2 log4j.appender.console=org.apache.log4j.ConsoleAppender log4j.appende

配置log4j和logback將日誌記錄到syslog

1. log4j配置內容如下 log4j.appender.syslog=org.apache.log4j.net.SyslogAppender log4j.appender.syslog.SyslogHost=127.0.0.1 log4j.appender.syslog

使程式在後臺執行，並將日誌輸出至檔案

怎麼樣使程式在後臺執行///////////////////  nohup  ./nn > nn.log  2 > &1  &   ////////////方法有很多，這裡主

log4j2配置按照日誌級別將日誌輸出到不同的文件

hold eve open ror .com 可能 資料 clas opener 背景 在項目中，可能會產生非常多的日誌記錄，為了方便日誌分析，可以將日誌按級別輸出到指定文件。 log4j2.xml配置文件 <!--將info級別的日誌

ASP.NET Core 2.1 : 十二.內置日誌、使用Nlog將日誌輸出到文件（轉）

layout short including 文件名 rdquo 1.0 復雜 net 當前 應用離不開日誌，雖然現在使用VS有強大的調試功能，開發過程中不復雜的情況懶得輸出日誌了（想起print和echo的有木有），但在一些復雜的過程中以及應用日常運行中的日誌還是非常有用

Log4j輸出日誌到kafka

一、新增需要的jar包        去掉springboot專案自帶的日誌jar包，新增kafka和log4j的jar包 <dependency> <groupId>org.springframework.boot</

實時分析Flume-Kafka框架搭建最終將資料在mysql中輸出

因為搭建框架比較複雜如果這其中有不足，歡迎提出指正。下面附上實時分析簡化框架圖幫助理解。 把離線分析框架也附上 實時分析搭建過程： 1.在命令提示符中（Windows+R）找到準備好的SocketTest.java路徑，javac SocketTest.java執行後生

SpringBoot入門系列之第二篇:logback日誌輸出

日誌在ApplicationContext建立之前載入。 所以不能通過在@Configuration檔案中的@PropertySources來進行配置 Logback：logback-spring.xml, logback-spring.groovy, logback.xml or logback.groov

Kafka專案實戰－使用者日誌上報實時統計之編碼實踐

1.概述 該課程我以使用者實時上報日誌案例為基礎，帶著大家去完成各個KPI的編碼工作，實現生產模組、消費模組，資料持久化，以及應用排程等工作， 通過對這一系列流程的演示，讓大家能夠去掌握Kafka專案的相關編碼以及排程流程。下面，我們首先來預覽本課程所包含的課時，他們分別