消息系统kafka介绍

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，这样可以实时的观测到最新的消息。

 

• test-kafka.zip (14.7 KB)
• 下载次数: 22

评论

3 楼 西巴拉古呀那 2018-02-21  

Kafka分布式消息系统实战（与JavaScalaHadoopStorm集成）
网盘地址：https://pan.baidu.com/s/1nwwhpP3 密码: mxu6
网盘地址：https://pan.baidu.com/s/1mjM5HaC 密码: xa5s

2 楼 kafodaote 2018-01-21  

Kafka分布式消息系统实战（与JavaScalaHadoopStorm集成）
网盘地址：https://pan.baidu.com/s/1c3JymAk 密码: dnky
网盘地址：https://pan.baidu.com/s/1eTV5ygU 密码: 3g3v

1 楼 成大大的 2017-10-18  

Kafka分布式消息系统实战（与JavaScalaHadoopStorm集成）
——https://pan.baidu.com/s/1cAm9AI 密码: 7fvt

内容简介
Kafka是分布式的消息队列，作为云计算服务的基石，它广泛的应用在实时数据流方面，是实时数据处理的数据中枢，广泛应用在很多互联网企业，例如：linkedin，facebook，腾讯，百度，阿里等。实时数据流是现在互联网公司、甚至拥有大规模数据的传统企业的主要模式， 实时数据(Real-time Activity Data)就是那些非交易，不需要秒级响应的数据， 但在后续的分析中产生极大作用，例如个性化推荐、运营服务监控、精细化营销、报表等 。

本课程的目的在于系统性地介绍Kafka分布式消息系统，掌握了Kafka，你就拿到了大数据处理领域消息处理机制的钥匙，能够轻松上手开发分布式消息系统应用程序开发和维护，笑傲大数据处理技术。学完本课程，你可以掌握：

1.Kafka的部署方式
2.Kafka的原理
3.Kafka与其他大数据组件的集成
4.基于Kafka的程序开发



第一章　Kafka的基本介绍
   1.1 什么是消息系统
   1.2 消息队列的分类
   1.3 Kafka的基本架构和概念
   1.4 ZooKeeper简介和安装

第二章　Kafka的原理解析
   2.1 Kafka的Producer处理逻辑
   2.2 Kafka的broker处理逻辑
   2.3 Kafka的Consumer处理逻辑
   2.4 Kafka集群部署在zk里的存储结构

第三章　Kafka的部署方式
   3.1 Kafka伪集群部署模式
   3.2 Kafka集群部署模式
   3.3 核心配置文件server.properties

第四章 Kafka的Java应用开发
   4.1 Producer端的实现
   4.2 Consumer端的实现

第五章 Kafka的Scala应用开发
   5.1 Scala的Producer的实现
   5.2 Scala的Producer的实现

第六章　Kafka与Hadoop的集成
   6.1 Hadoop简介和配置
   6.2 集成Kafka和Hadoop

第七章　Kafka与Flume的集成
   7.1 Flume简介和使用
   7.2 集成Kafka和Flume

第八章　Kafka与Storm的集成
   8.1 Storm的运行机制和部署
   8.2 Storm编程案例
   8.3 集成Kafka和Storm