(一个)kafka-jstorm集群实时日志分析 它 ———kafka实时日志处理

(一个)kafka-jstorm集群实时日志分析 它 ———kafka实时日志处理

大家好,又见面了,我是全栈君,今天给大家准备了Idea注册码。

package com.doctor.logbackextend;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;

import org.apache.commons.lang.RandomStringUtils;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * zookeeper 和kafka环境准备好。

本地端口号默认设置 * * @author doctor * * @time 2014年10月24日 下午3:14:01 */public class KafkaAppenderTest { private static final Logger LOG = LoggerFactory.getLogger(KafkaAppenderTest.class); /** 先启动此測试方法,模拟log日志输出到kafka */ @Test public void test_log_producer() { while(true){ LOG.info("test_log_producer : " + RandomStringUtils.random(3, "hello doctro,how are you,and you")); } } /** 再启动此測试方法。模拟消费者获取日志,进而分析,此方法不过打印打控制台,不是log。防止模拟log測试方法数据混淆 */ @Test public void test_comsumer(){ Properties props = new Properties(); props.put("zookeeper.connect", "127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183"); props.put("group.id", "kafkatest-group");// props.put("zookeeper.session.timeout.ms", "400");// props.put("zookeeper.sync.time.ms", "200");// props.put("auto.commit.interval.ms", "1000"); ConsumerConfig paramConsumerConfig = new ConsumerConfig(props ); ConsumerConnector consumer = Consumer.createJavaConsumerConnector(paramConsumerConfig ); Map<String, Integer> topicCountMap = new HashMap<>(); topicCountMap.put("kafka-test", new Integer(1)); Map<String, List<KafkaStream<byte[], byte[]>>> consumerStream = consumer.createMessageStreams(topicCountMap); List<KafkaStream<byte[], byte[]>> streams = consumerStream.get("kafka-test"); for (KafkaStream<byte[], byte[]> stream : streams) { ConsumerIterator<byte[], byte[]> it = stream.iterator(); while(it.hasNext()) System.out.println(new String("test_comsumer: " + new String(it.next().message()))); } }}

      为了实时日志处理互联网系统的日志,对于电商来说具有非常重要的意义,比方,淘宝购物时候,你浏览某些商品的时候。系统后台实时日志处理分析后,系统能够向用户实时推荐给用户相关商品。来引导用户的选择等等。

        为了实时日志处理。我们选择kafka集群,日志的处理分析选择jstorm集群,至于jstorm处理的结果,你能够选择保存到数据库里。入hbase、mysql。maridb等。

系统的日志接口选择了slf4j,logback组合,为了让系统的日志可以写入kafka集群,选择扩展logback Appender。在logback里配置一下。就行自己主动输出日志到kafka集群。

kafka的集群安装,在此不介绍了,为了模拟真实性,zookeeper本地集群也安装部署了。


以下是怎样扩展logback Appender

package com.doctor.logbackextend;

import java.util.Properties;

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

public class KafkaAppender extends AppenderBase<ILoggingEvent> {

	private String topic;
	private String zookeeperHost;
	

	private String broker;
	private Producer<String, String> producer;
	private Formatter formatter;
	
	public String getBroker() {
		return broker;
	}

	public void setBroker(String broker) {
		this.broker = broker;
	}
	@Override
	protected void append(ILoggingEvent eventObject) {
		String message = this.formatter.formate(eventObject);
		this.producer.send(new KeyedMessage<String, String>(this.topic, message));

	}

	@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("metadata.broker.list", this.broker);
		props.put("serializer.class", "kafka.serializer.StringEncoder");
		
		ProducerConfig config = new ProducerConfig(props);
		this.producer = new Producer<String, String>(config);
	}

	@Override
	public void stop() {
		super.stop();
		this.producer.close();
	}

	
	
	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 Producer<String, String> getProducer() {
		return producer;
	}

	public void setProducer(Producer<String, String> producer) {
		this.producer = producer;
	}


	public Formatter getFormatter() {
		return formatter;
	}

	public void setFormatter(Formatter formatter) {
		this.formatter = formatter;
	}
	
	
	
	/**
	 * 格式化日志格式
	 * @author doctor
	 *
	 * @time   2014年10月24日 上午10:37:17
	 */
	interface Formatter{
		String formate(ILoggingEvent event);
	}
	
	public static class MessageFormatter implements Formatter{

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


对于日志的输出格式
MessageFormatter没有特殊处理,由于仅仅是模拟一下,你能够制定你的格式,入json等。

在logback.xml的配置例如以下:

<appender name="kafka" class="com.doctor.logbackextend.KafkaAppender">
 		<topic>kafka-test</topic>
 		<!-- <zookeeperHost>127.0.0.1:2181</zookeeperHost> -->
 		<!-- <broker>127.0.0.1:9092</broker> -->
 		<zookeeperHost>127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183</zookeeperHost>
 		<broker>127.0.0.1:9092,127.0.0.1:9093</broker>
 	</appender>
 	
 	
	<root level="all">
		<appender-ref ref="stdout" />
		<appender-ref ref="defaultAppender" />
		<appender-ref ref="kafka" />
	</root>

  <zookeeperHost>

    我本地启动了三个zookeer。依据配置。你能够知道是怎样配置的吧。

   kafka集群的broker我配置了两个,都是在本地机器。


測试代码:

package com.doctor.logbackextend;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;

import org.apache.commons.lang.RandomStringUtils;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * zookeeper 和kafka环境准备好。本地端口号默认设置
 * 
 * @author doctor
 *
 * @time   2014年10月24日 下午3:14:01
 */
public class KafkaAppenderTest {
	private static final Logger LOG = LoggerFactory.getLogger(KafkaAppenderTest.class);
	

	/** 先启动此測试方法,模拟log日志输出到kafka */
	@Test
	public void test_log_producer() {
		while(true){
			LOG.info("test_log_producer : "  + RandomStringUtils.random(3, "hello doctro,how are you,and you"));
		}
	}
	
	
	/** 再启动此測试方法,模拟消费者获取日志,进而分析,此方法不过打印打控制台,不是log。防止模拟log測试方法数据混淆 */
	@Test
	public void test_comsumer(){
		Properties props = new Properties();
		props.put("zookeeper.connect", "127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183");
		props.put("group.id", "kafkatest-group");
//		props.put("zookeeper.session.timeout.ms", "400");
//		props.put("zookeeper.sync.time.ms", "200");
//		props.put("auto.commit.interval.ms", "1000");
		ConsumerConfig paramConsumerConfig = new ConsumerConfig(props );
		ConsumerConnector consumer = Consumer.createJavaConsumerConnector(paramConsumerConfig );
		
		Map<String, Integer> topicCountMap = new HashMap<>();
		topicCountMap.put("kafka-test", new Integer(1));
		Map<String, List<KafkaStream<byte[], byte[]>>> consumerStream = consumer.createMessageStreams(topicCountMap);
		List<KafkaStream<byte[], byte[]>> streams = consumerStream.get("kafka-test");
		
		for (KafkaStream<byte[], byte[]> stream : streams) {
			ConsumerIterator<byte[], byte[]> it = stream.iterator();
			while(it.hasNext())
			System.out.println(new String("test_comsumer: " + new String(it.next().message())));
		}
		
		
	}

}


结果,明天再附上截图。

版权声明:本文博客原创文章,博客,未经同意,不得转载。

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。

发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/117693.html原文链接:https://javaforall.cn

【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛

【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...

(0)


相关推荐

  • 浅谈 MVC与三层架构

    浅谈 MVC与三层架构引言:使用Eclipse开发工具写JavaWeb项目时会发现,一个中型或者大型项目随着代码的增多,会发现:代码既可以写在src目录下,也可以写在WebContent目录下。src下可以建很多包,WebContent下可以建很多文件夹。所以问题就来了:一个新的类到底往哪个目录下的哪个文件夹里写?此时解决办法就是:需要一个模式去规范,到底哪个类该往哪里写。…

  • JDBC连接数据库的步骤

    JDBC连接数据库的步骤JDBC连接数据库一共有7步。1、首先加载驱动2、提供JDBC连接的URL3、创建数据库的连接4、创建一个statement执行者5、执行SQL语句6、处理返回结果7、关闭JDBC对象

  • 交换机基础配置教程[通俗易懂]

    交换机基础配置教程[通俗易懂]一、带外管理网络的管理控制信息与用户网络的承载业务信息在不同的逻辑信道传送,也就是设备提供专门用于管理的宽带console口:专门用来管理的,并不传输数据,接入一根console线,是一个扁平电缆,另外一端是一个串行接口,用来接入电脑或笔记本上,近端管理时我们会使用console线进行近端管理,设备要做密码恢复时,时必须要进行近端管理的,只有通过console空才能进行密码恢复。以太网口:以太网口用来传输数据管理信息和数据传输是隔离的,所以我们称之为带外管理初次配置:通过console口进行配置,需

  • 通过DatabaseMetaData分析当前Connnection连接的数据库信息

    通过DatabaseMetaData分析当前Connnection连接的数据库信息packagejavaxxx;importjava.io.FileInputStream;importjava.io.FileNotFoundException;importjava.io.IOException;importjava.io.InputStream;importjava.sql.Connection;importjava.sql.DatabaseMeta…

  • 串口调试助手(CM野人版)4.0有严重Bug,已经跟作者反应,等候更新

    串口调试助手(CM野人版)4.0有严重Bug,已经跟作者反应,等候更新这个软件,看起来很漂亮,能任意设置波特率,支持921600波特率,看起来很华丽。但最近发现其一个严重的Bug:版本就是4.0我打印了一些信息:能正常显示。我把这些打印的代码行给注释了,依然能打印出来,见鬼了。然后我重启电脑,问题消失,不再打印注释过的代码行。我换了别的串口助手,问题消失。所以一定是这个助手的问题。基础的调试软件出问题,非常不易发现,带来的损失非常大。建

  • mac键盘快捷键使用大全_苹果电脑shift是哪个键

    mac键盘快捷键使用大全_苹果电脑shift是哪个键为了让初次用Mac的新手和小白们更快地上手,小编整理了一份Mac常用快捷键列表,虽然都是比较基础的入门知识,但也能帮助到大家更快地学习和查找macOS的键盘快捷键,以此提高电脑的操作效率。Mac与Windows的几个不同按键一切开始前,我们先来认识一下苹果Mac键盘上几个陌生的按键,比如⌘(Command/Cmd)、⌥“Option”、⌃“Ctrl”、⇧“Shift”等等。其中最为独特的就是Mac键盘专有的按键⌘,它叫做“Command”键(缩写叫Cmd,

发表回复

您的电子邮箱地址不会被公开。

关注全栈程序员社区公众号