Kafka之——扩展logback将日志输出到Kafka实例
程序员文章站
2024-01-11 20:58:40
...
转载请注明出处：http://blog.csdn.net/l1028386804/article/details/79135948
一、Kafka的安装
请参见博文《Kafka之——安装并（单节点）配置启动Kafka》、《Kafka之——单机多broker实例集群搭建》、《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类，可以看到打印出如下日志：
[aaa@qq.com 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).四、温馨提示
大家可以到链接http://download.csdn.net/download/l1028386804/10219010下载Kafka扩展logback将日志输出到Kafka实例源码
上一篇： 前端学PHP之基础语法，前端php语法
下一篇： docker部署jar