http://dongxicheng.org/search-engine/kafka/
http://shift-alt-ctrl.iteye.com/blog/1930791
kafka作为分布式日志收集或系统监控服务,我们有必要在合适的场合使用它。kafka的部署包括zookeeper环境/kafka环境,同时还需要进行一些配置操作.接下来介绍如何使用kafka.
我们使用3个zookeeper实例构建zk集群,使用2个kafka broker构建kafka集群.
其中kafka为0.8V,zookeeper为3.4.5V
一.Zookeeper集群构建
我们有3个zk实例,分别为zk-0,zk-1,zk-2;如果你仅仅是测试使用,可以使用1个zk实例.
1) zk-0
调整配置文件:
- clientPort=2181
- server.0=127.0.0.1:2888:3888
- server.1=127.0.0.1:2889:3889
- server.2=127.0.0.1:2890:3890
- ##只需要修改上述配置,其他配置保留默认值
启动zookeeper
- ./zkServer.sh start
2) zk-1
调整配置文件(其他配置和zk-0一只):
- clientPort=2182
- ##只需要修改上述配置,其他配置保留默认值
启动zookeeper
- ./zkServer.sh start
3) zk-2
调整配置文件(其他配置和zk-0一只):
- clientPort=2183
- ##只需要修改上述配置,其他配置保留默认值
启动zookeeper
- ./zkServer.sh start
二. Kafka集群构建
因为Broker配置文件涉及到zookeeper的相关约定,因此我们先展示broker配置文件.我们使用2个kafka broker来构建这个集群环境,分别为kafka-0,kafka-1.
1) kafka-0
在config目录下修改配置文件为:
- broker.id=0
- port=9092
- num.network.threads=2
- num.io.threads=2
- socket.send.buffer.bytes=1048576
- socket.receive.buffer.bytes=1048576
- socket.request.max.bytes=104857600
- log.dir=./logs
- num.partitions=2
- log.flush.interval.messages=10000
- log.flush.interval.ms=1000
- log.retention.hours=168
- #log.retention.bytes=1073741824
- log.segment.bytes=536870912
- num.replica.fetchers=2
- log.cleanup.interval.mins=10
- zookeeper.connect=127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183
- zookeeper.connection.timeout.ms=1000000
- kafka.metrics.polling.interval.secs=5
- kafka.metrics.reporters=kafka.metrics.KafkaCSVMetricsReporter
- kafka.csv.metrics.dir=/tmp/kafka_metrics
- kafka.csv.metrics.reporter.enabled=false
因为kafka用scala语言编写,因此运行kafka需要首先准备scala相关环境。
- > cd kafka-0
- > ./sbt update
- > ./sbt package
- > ./sbt assembly-package-dependency
其中最后一条指令执行有可能出现异常,暂且不管。 启动kafka broker:
- > JMS_PORT=9997 bin/kafka-server-start.sh config/server.properties &
因为zookeeper环境已经正常运行了,我们无需通过kafka来挂载启动zookeeper.如果你的一台机器上部署了多个kafka broker,你需要声明JMS_PORT.
2) kafka-1
- broker.id=1
- port=9093
- ##其他配置和kafka-0保持一致
然后和kafka-0一样执行打包命令,然后启动此broker.
- > JMS_PORT=9998 bin/kafka-server-start.sh config/server.properties &
到目前为止环境已经OK了,那我们就开始展示编程实例吧。
三.项目准备
项目基于maven构建,不得不说kafka java客户端实在是太糟糕了;构建环境会遇到很多麻烦。建议参考如下pom.xml;其中各个依赖包必须版本协调一致。如果kafka client的版本和kafka server的版本不一致,将会有很多异常,比如"broker id not exists"等;因为kafka从0.7升级到0.8之后(正名为2.8.0),client与server通讯的protocol已经改变.
- <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.test</groupId>
- <artifactId>test-kafka</artifactId>
- <packaging>jar</packaging>
- <name>test-kafka</name>
- <url>http://maven.apache.org</url>
- <version>1.0.0</version>
- <dependencies>
- <dependency>
- <groupId>log4j</groupId>
- <artifactId>log4j</artifactId>
- <version>1.2.14</version>
- </dependency>
- <dependency>
- <groupId>org.apache.kafka</groupId>
- <artifactId>kafka_2.8.0</artifactId>
- <version>0.8.0-beta1</version>
- <exclusions>
- <exclusion>
- <groupId>log4j</groupId>
- <artifactId>log4j</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
- <dependency>
- <groupId>org.scala-lang</groupId>
- <artifactId>scala-library</artifactId>
- <version>2.8.1</version>
- </dependency>
- <dependency>
- <groupId>com.yammer.metrics</groupId>
- <artifactId>metrics-core</artifactId>
- <version>2.2.0</version>
- </dependency>
- <dependency>
- <groupId>com.101tec</groupId>
- <artifactId>zkclient</artifactId>
- <version>0.3</version>
- </dependency>
- </dependencies>
- <build>
- <finalName>test-kafka-1.0</finalName>
- <resources>
- <resource>
- <directory>src/main/resources</directory>
- <filtering>true</filtering>
- </resource>
- </resources>
- <plugins>
- <plugin>
- <artifactId>maven-compiler-plugin</artifactId>
- <version>2.3.2</version>
- <configuration>
- <source>1.5</source>
- <target>1.5</target>
- <encoding>gb2312</encoding>
- </configuration>
- </plugin>
- <plugin>
- <artifactId>maven-resources-plugin</artifactId>
- <version>2.2</version>
- <configuration>
- <encoding>gbk</encoding>
- </configuration>
- </plugin>
- </plugins>
- </build>
- </project>
四.Producer端代码
1) producer.properties文件:此文件放在/resources目录下
- #partitioner.class=
- metadata.broker.list=127.0.0.1:9092,127.0.0.1:9093
- ##,127.0.0.1:9093
- producer.type=sync
- compression.codec=0
- serializer.class=kafka.serializer.StringEncoder
- ##在producer.type=async时有效
- #batch.num.messages=100
2) LogProducer.java代码样例
- package com.test.kafka;
- import java.util.ArrayList;
- import java.util.Collection;
- import java.util.List;
- import java.util.Properties;
- import kafka.javaapi.producer.Producer;
- import kafka.producer.KeyedMessage;
- import kafka.producer.ProducerConfig;
- public class LogProducer {
- private Producer<String,String> inner;
- public LogProducer() throws Exception{
- Properties properties = new Properties();
- properties.load(ClassLoader.getSystemResourceAsStream("producer.properties"));
- ProducerConfig config = new ProducerConfig(properties);
- inner = new Producer<String, String>(config);
- }
- public void send(String topicName,String message) {
- if(topicName == null || message == null){
- return;
- }
- KeyedMessage<String, String> km = new KeyedMessage<String, String>(topicName,message);
- inner.send(km);
- }
- public void send(String topicName,Collection<String> messages) {
- if(topicName == null || messages == null){
- return;
- }
- if(messages.isEmpty()){
- return;
- }
- List<KeyedMessage<String, String>> kms = new ArrayList<KeyedMessage<String, String>>();
- for(String entry : messages){
- KeyedMessage<String, String> km = new KeyedMessage<String, String>(topicName,entry);
- kms.add(km);
- }
- inner.send(kms);
- }
- public void close(){
- inner.close();
- }
- /**
- * @param args
- */
- public static void main(String[] args) {
- LogProducer producer = null;
- try{
- producer = new LogProducer();
- int i=0;
- while(true){
- producer.send("test-topic", "this is a sample" + i);
- i++;
- Thread.sleep(2000);
- }
- }catch(Exception e){
- e.printStackTrace();
- }finally{
- if(producer != null){
- producer.close();
- }
- }
- }
- }
五.Consumer端
1) consumer.properties:文件位于/resources目录下
- zookeeper.connect=127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183
- ##,127.0.0.1:2182,127.0.0.1:2183
- # timeout in ms for connecting to zookeeper
- zookeeper.connectiontimeout.ms=1000000
- #consumer group id
- group.id=test-group
- #consumer timeout
- #consumer.timeout.ms=5000
- auto.commit.enable=true
- auto.commit.interval.ms=60000
2) LogConsumer.java代码样例
- package com.test.kafka;
- import java.util.HashMap;
- import java.util.List;
- import java.util.Map;
- import java.util.Properties;
- import java.util.concurrent.ExecutorService;
- import java.util.concurrent.Executors;
- import kafka.consumer.Consumer;
- import kafka.consumer.ConsumerConfig;
- import kafka.consumer.ConsumerIterator;
- import kafka.consumer.KafkaStream;
- import kafka.javaapi.consumer.ConsumerConnector;
- import kafka.message.MessageAndMetadata;
- public class LogConsumer {
- private ConsumerConfig config;
- private String topic;
- private int partitionsNum;
- private MessageExecutor executor;
- private ConsumerConnector connector;
- private ExecutorService threadPool;
- public LogConsumer(String topic,int partitionsNum,MessageExecutor executor) throws Exception{
- Properties properties = new Properties();
- properties.load(ClassLoader.getSystemResourceAsStream("consumer.properties"));
- config = new ConsumerConfig(properties);
- this.topic = topic;
- this.partitionsNum = partitionsNum;
- this.executor = executor;
- }
- public void start() throws Exception{
- connector = Consumer.createJavaConsumerConnector(config);
- Map<String,Integer> topics = new HashMap<String,Integer>();
- topics.put(topic, partitionsNum);
- Map<String, List<KafkaStream<byte[], byte[]>>> streams = connector.createMessageStreams(topics);
- List<KafkaStream<byte[], byte[]>> partitions = streams.get(topic);
- threadPool = Executors.newFixedThreadPool(partitionsNum);
- for(KafkaStream<byte[], byte[]> partition : partitions){
- threadPool.execute(new MessageRunner(partition));
- }
- }
- public void close(){
- try{
- threadPool.shutdownNow();
- }catch(Exception e){
- //
- }finally{
- connector.shutdown();
- }
- }
- class MessageRunner implements Runnable{
- private KafkaStream<byte[], byte[]> partition;
- MessageRunner(KafkaStream<byte[], byte[]> partition) {
- this.partition = partition;
- }
- public void run(){
- ConsumerIterator<byte[], byte[]> it = partition.iterator();
- while(it.hasNext()){
- //connector.commitOffsets();手动提交offset,当autocommit.enable=false时使用
- MessageAndMetadata<byte[],byte[]> item = it.next();
- System.out.println("partiton:" + item.partition());
- System.out.println("offset:" + item.offset());
- executor.execute(new String(item.message()));//UTF-8,注意异常
- }
- }
- }
- interface MessageExecutor {
- public void execute(String message);
- }
- /**
- * @param args
- */
- public static void main(String[] args) {
- LogConsumer consumer = null;
- try{
- MessageExecutor executor = new MessageExecutor() {
- public void execute(String message) {
- System.out.println(message);
- }
- };
- consumer = new LogConsumer("test-topic", 2, executor);
- consumer.start();
- }catch(Exception e){
- e.printStackTrace();
- }finally{
- // if(consumer != null){
- // consumer.close();
- // }
- }
- }
- }
需要提醒的是,上述LogConsumer类中,没有太多的关注异常情况,必须在MessageExecutor.execute()方法中抛出异常时的情况.
在测试时,建议优先启动consumer,然后再启动producer,这样可以实时的观测到最新的消息。
相关推荐
Kafka是大数据平台中数据缓存的中间件,资源里有一篇关于Kafka原作者的一篇论文,讲解的很详细。也有两份介绍Kafka使用的PPT,还有Kafka的源代码。
Kafka是最初由Linkedin公司开发,是一个分布式、分区的、多副本的、多订阅者,基于zookeeper协调的分布式日志系统(也可以当做MQ系统),常见可以用于web/nginx日志、访问日志,消息服务等等,Linkedin于2010年贡献...
Kafka是Linkedin于2010年12月份开源的消息系统,它主要用于处理活跃的流式数据和运营数据,同时也是支持通用的消息语义(messaging semantics)。其中活跃的流式数据包括页面访问量(page view)、被查看内容方面的...
消息队列降低了进程间的耦合度,所以即使一个处理消息的进程挂掉,加入队列中的消息仍然可以在系统恢复后被处理。 顺序保证 在大多使用场景下,数据处理的顺序都很重要。大部分消息队列本来就是排序的,并且能保证...
1.1. 为何使用消息系统 1.1.1. 解耦 在项目启动之初来预测将来项目会碰到什么需求, 是极其困难的。 消息系统在处理过程中间插入了一个隐含的、 基于数据的接口层, 两边的处理过程都要实现这一接口。 这允许你独立...
本文详细解析了Kafka的设计原理,重点介绍了Kafka作为一种高效的分布式消息系统的核心组件和机制。首先,文档解释了Kafka的总控制器(Controller)的作用,它负责管理集群中的分区和副本状态,并在必要时进行Leader...
Kafka是一种高吞吐量的分布式发布订阅消息系统,它可以处理消费者在网站中的所有动作流数据。 这种动作(网页浏览,搜索和其他用户的行动)是在现代网络上的许多社会功能的一个关键因素。 这些数据通常是由于吞吐量...
介绍Kafka背景,使用消息系统的优势,常用消息系统对比,Kafka架构介绍,Kafka实现语义分析,Replication及Leader Election机制剖析,Consumer Group Rebalance实现原理介绍,以及Benchmark测试。
kafka介绍:Kafka知识领域+消息队列技术关键词+安装、配置、使用、主题、分区、副本、消费者组、流处理、事件源存储、性能优化、安全性、集群部署内容关键词+构建实时数据管道、解耦服务、可靠消息传递、数据可靠性...
kafka消息系统的设计介绍,转载对kafka系统设计的翻译。
Kafka 入门、Kafka 的基本使用、客户端详解、Kafka 原理介绍、Kafka 运维与监控以及高级 Kafka 应用,在实际业务系统中实现消息队列应用、应用程序集成、分布式存储构建,甚至是流处理应用的开发与部署。
消息队列中间件是分布式系统中重要的组件,主要解决应用解耦,异步消息,流量削锋等问题,实现高性能,高可用,可伸缩和最终一致性架构。目前使用较多的消息队列有ActiveMQ,RabbitMQ,ZeroMQ,Kafka,MetaMQ,...
本文介绍如何在 Spring Boot 3.X 中利用 Kafka 实现高效的消息传递功能。Kafka 是一个分布式流处理平台,适用于实时数据流处理、日志收集与分析、事件驱动等场景。通过集成 Kafka,我们可以实现可靠的异步消息传递,...
本文介绍了Kafka的创建背景,设计目标,使用消息系统的优势以及目前流行的消息系统对比。并介绍了Kafka的架构,Producer消息路由,ConsumerGroup以及由其实现的不同消息分发方式,Topic&Partition,最后介绍了Kafka...
本文介绍如何在 Spring Boot 3.X 中利用 Kafka 实现高效的消息传递功能。Kafka 是一个分布式流处理平台,适用于实时数据流处理、日志收集与分析、事件驱动等场景。通过集成 Kafka,我们可以实现可靠的异步消息传递,...
本文介绍如何在 Spring Boot 3.X 中利用 Kafka 实现高效的消息传递功能。Kafka 是一个分布式流处理平台,适用于实时数据流处理、日志收集与分析、事件驱动等场景。通过集成 Kafka,我们可以实现可靠的异步消息传递,...
本文介绍如何在 Spring Boot 3.X 中利用 Kafka 实现高效的消息传递功能。Kafka 是一个分布式流处理平台,适用于实时数据流处理、日志收集与分析、事件驱动等场景。通过集成 Kafka,我们可以实现可靠的异步消息传递,...
《Apache Kafka实战》共分为10章:第1章全面介绍消息引擎系统以及Kafka的基本概念与特性,快速带领读者走进Kafka的世界;第2章简要回顾了Apache Kafka的发展历史;第3章详细介绍了Kafka集群环境的搭建;第4、5章深入...
本文档提供了对Kafka这一分布式消息系统的全面解析,从基本概念到实际应用,涵盖了其在日志收集、消息系统、用户活动跟踪等方面的使用场景。首先介绍了Kafka的核心概念,如Broker、Topic、Producer、Consumer等,...
Kafka是一种高吞吐量的分布式发布订阅消息系统,它可以处理消费者规模的网站中的所有动作流数据。Kafka的目的是通过Hadoop的并行加载机制来统一线上和离线的消息处理,也是为了通过集群机来提供实时的消费。文档介绍...