Kafka集群的安裝和使用

（有點像流水賬，好記性不如爛筆頭，權記於此以備忘）

總結：

概念：Broker、Producer、Consumer、Consumer Group、Topic、Partition、Replication

特點：分布式、高可用、高吞吐量、數據只在Partition內有序等

高寫性能的原因：分Partition以支持高並發讀寫，順序追加寫

 

Kafka是一種高吞吐量的分布式發布訂閱的消息隊列系統，原本開發自LinkedIn，用作LinkedIn的活動流（ActivityStream）和運營數據處理管道（Pipeline）的基礎。現在它已被多家不同類型的公司作為多種類型的數據管道和消息系統使用。

1 Kafka消息隊列簡介

1.1 基本術語

• Broker
  Kafka集群包含一個或多個服務器，這種服務器被稱為broker[5]  
• Topic
  每條發布到Kafka集群的消息都有一個類別，這個類別被稱為Topic。（物理上不同Topic的消息分開存儲，邏輯上一個Topic的消息雖然保存於一個或多個broker上但用戶只需指定消息的Topic即可生產或消費數據而不必關心數據存於何處）
• Partition
  Partition是物理上的概念，每個Topic包含一個或多個Partition.
• Producer
  負責發布消息到Kafka broker
• Consumer
  消息消費者，向Kafka broker讀取消息的客戶端。
• Consumer Group
  每個Consumer屬於一個特定的Consumer Group（可為每個Consumer指定group name，若不指定group name則屬於默認的group）。

1.2 消息隊列

1.2.1 基本特性

1. 可擴展
   
   • 在不需要下線的情況下進行擴容
   • 數據流分區(partition)存儲在多個機器上
2. 高性能
   
   • 單個broker就能服務上千客戶端
   • 單個broker每秒種讀/寫可達每秒幾百兆字節
   • 多個brokers組成的集群將達到非常強的吞吐能力
   • 性能穩定，無論數據多大
   • Kafka在底層摒棄了Java堆緩存機制，采用了操作系統級別的頁緩存，同時將隨機寫操作改為順序寫，再結合Zero-Copy的特性極大地改善了IO性能。
3. 持久存儲
   
   • 存儲在磁盤上
   • 冗余備份到其他服務器上以防止丟失

1.2.2 消息格式

1. 一個topic對應一種消息格式，因此消息用topic分類
2. 一個topic代表的消息有1個或者多個patition(s)組成
3. 一個partition中
   
   • 一個partition應該存放在一到多個server上
     • 如果只有一個server，就沒有冗余備份，是單機而不是集群
     • 如果有多個server
       • 一個server為leader，其他servers為followers；leader需要接受讀寫請求；followers僅作冗余備份；leader出現故障，會自動選舉一個follower作為leader，保證服務不中斷；每個server都可能扮演一些partitions的leader和其它partitions的follower角色，這樣整個集群就會達到負載均衡的效果
   • 消息按順序存放，順序不可變
   • 只能追加消息，不能插入
   • 每個消息都有一個offset，用作消息ID, 在一個partition中唯一
   • offset由consumer保存和管理，因此讀取順序實際上是完全有consumer決定的，不一定是線性的
   • 消息有超時日期，過期則刪除

1.2.3 生產者 producer

• producer將消息寫入kafka
• 寫入要指定topic和partition
• 消息如何分到不同的partition，算法由producer指定

1.2.4 消費者 consumer

• consumer讀取消息並作處理
• consumer group
  • 這個概念的引入為了支持兩種場景：每條消息分發一個消費者，每條消息廣播給消費組的所有消費者
  • 多個consumer group訂閱一個topic，該topci的消息廣播給所有consumer group
  • 一條消息發送到一個consumer group后，只能由該group的一個consumer接收和使用
  • 一個group中的每個consumer各自對應一個partition可以帶來如下好處
    • 可以按照partition的數目進行並發處理
    • 每個partition都只有一個consumer讀取，因而保證了消息被處理的順序是按照partition的存放順序進行，注意這個順序受到producer存放消息的算法影響

•  一個Consumer可以有多個線程進行消費，線程數應不多於topic的partition數，因為對於一個包含一或多消費線程的consumer group來說，一個partition只能分給其中的一個消費線程消費，且讓盡可能多的線程能分配到partition（不過實際上真正去消費的線程及線程數還是由線程池的調度機制來決定）。這樣如果線程數比partition數多，那么單射分配也會有多出的線程，它們就不會消費到任何一個partition的數據而空轉耗資源 。
• 如果consumer從多個partition讀到數據，不保證數據間的順序性，kafka只保證在一個partition上數據是有序的，但多個partition，根據你讀的順序會有不同
• 增減consumer，broker，partition會導致rebalance，所以rebalance后consumer對應的partition會發生變化

2. 安裝和使用

以kafka_2.11-0.10.0.0為例。

下載解壓后，進入kafka_2.11-0.10.0.0/

2.1 啟動Zookeeper

測試時可以使用Kafka附帶的Zookeeper：

啟動： ./bin/zookeeper-server-start.sh config/zookeeper.properties & ，config/zookeeper.properties是Zookeeper的配置文件。

結束： ./bin/zookeeper-server-stop.sh 

不過最好自己搭建一個Zookeeper集群，提高可用性和可靠性。詳見：Zookeeper的安裝和使用——MarchOn

2.2 啟動Kafka服務器

2.2.1 配置文件

配置config/server.properties文件，一般需要配置如下字段，其他按默認即可：

broker.id：    　　　　　　每一個broker在集群中的唯一表示，要求是正數
listeners（效果同之前的版本的host.name及port）：注意綁定host.name，否則可能出現莫名其妙的錯誤如consumer找不到broker。這個host.name是Kafka的server的機器名字，會注冊到Zookeeper中
log.dirs：    　　　　　　 kafka數據的存放地址，多個地址的話用逗號分割,多個目錄分布在不同磁盤上可以提高讀寫性能
log.retention.hours：   　數據文件保留多長時間， 存儲的最大時間超過這個時間會根據log.cleanup.policy設置數據清除策略
zookeeper.connect：   　　指定ZooKeeper的connect string，以hostname:port的形式，可有多個以逗號分隔，如hostname1:port1,hostname2:port2,hostname3:port3，還可有路徑，如：hostname1:port1,hostname2:port2,hostname3:port3/kafka，注意要事先在zk中創建/kafka節點，否則會報出錯誤：java.lang.IllegalArgumentException: Path length must be > 0

所有參數的含義及配置可參考：http://orchome.com/12、http://blog.csdn.net/lizhitao/article/details/25667831

 一個配置示例如下：

# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# see kafka.server.KafkaConfig for additional details and defaults

############################# Server Basics #############################

# The id of the broker. This must be set to a unique integer for each broker.
broker.id=1

############################# Socket Server Settings #############################

# The address the socket server listens on. It will get the value returned from 
# java.net.InetAddress.getCanonicalHostName() if not configured.
#   FORMAT:
#     listeners = security_protocol://host_name:port
#   EXAMPLE:
#     listeners = PLAINTEXT://your.host.name:9092
listeners=PLAINTEXT://192.168.6.128:9092

# Hostname and port the broker will advertise to producers and consumers. If not set, 
# it uses the value for "listeners" if configured.  Otherwise, it will use the value
# returned from java.net.InetAddress.getCanonicalHostName().
#advertised.listeners=PLAINTEXT://your.host.name:9092

# The number of threads handling network requests
num.network.threads=3

# The number of threads doing disk I/O
num.io.threads=8

# The send buffer (SO_SNDBUF) used by the socket server
socket.send.buffer.bytes=102400

# The receive buffer (SO_RCVBUF) used by the socket server
socket.receive.buffer.bytes=102400

# The maximum size of a request that the socket server will accept (protection against OOM)
socket.request.max.bytes=104857600


############################# Log Basics #############################

# A comma seperated list of directories under which to store log files
log.dirs=/usr/local/kafka/kafka_2.11-0.10.0.0/kfk_data/

# The default number of log partitions per topic. More partitions allow greater
# parallelism for consumption, but this will also result in more files across
# the brokers.
num.partitions=2
auto.create.topics.enable=false

# The number of threads per data directory to be used for log recovery at startup and flushing at shutdown.
# This value is recommended to be increased for installations with data dirs located in RAID array.
num.recovery.threads.per.data.dir=1

############################# Log Flush Policy #############################

# Messages are immediately written to the filesystem but by default we only fsync() to sync
# the OS cache lazily. The following configurations control the flush of data to disk.
# There are a few important trade-offs here:
#    1. Durability: Unflushed data may be lost if you are not using replication.
#    2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush.
#    3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to exceessive seeks.
# The settings below allow one to configure the flush policy to flush data after a period of time or
# every N messages (or both). This can be done globally and overridden on a per-topic basis.

# The number of messages to accept before forcing a flush of data to disk
#log.flush.interval.messages=10000

# The maximum amount of time a message can sit in a log before we force a flush
#log.flush.interval.ms=1000

############################# Log Retention Policy #############################

# The following configurations control the disposal of log segments. The policy can
# be set to delete segments after a period of time, or after a given size has accumulated.
# A segment will be deleted whenever *either* of these criteria are met. Deletion always happens
# from the end of the log.

# The minimum age of a log file to be eligible for deletion
log.retention.hours=4

# A size-based retention policy for logs. Segments are pruned from the log as long as the remaining
# segments don't drop below log.retention.bytes.
#log.retention.bytes=1073741824

# The maximum size of a log segment file. When this size is reached a new log segment will be created.
log.segment.bytes=1073741824

# The interval at which log segments are checked to see if they can be deleted according
# to the retention policies
log.retention.check.interval.ms=300000

############################# Zookeeper #############################

# Zookeeper connection string (see zookeeper docs for details).
# This is a comma separated host:port pairs, each corresponding to a zk
# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
# You can also append an optional chroot string to the urls to specify the
# root directory for all kafka znodes.
zookeeper.connect=192.168.6.131:2181,192.168.6.132:2181,192.168.6.133:2181

# Timeout in ms for connecting to zookeeper
zookeeper.connection.timeout.ms=6000

注意auto.create.topics.enable字段，若為true則如果producer寫入某個不存在的topic時會自動創建該topic，若為false則需要事先創建否則會報錯：failed after 3 retries。

2.2.2 命令

啟動： bin/kafka-server-start.sh config/server.properties ，生產環境最好以守護程序啟動：nohup  &

結束： bin/kafka-server-stop.sh 

2.2.3 Kafka在Zookeeper中的存儲結構

若上述的zookeeper.connect的值沒有路徑，則為根路徑，啟動Zookeeper和Kafka，命令行連接Zookeeper后，用 get / 命令可發現有 consumers、config、controller、admin、brokers、zookeeper、controller_epoch 這幾個目錄。

其結構如下：（具體可參考：apache kafka系列之在zookeeper中存儲結構）

 

2.3 使用

kafka本身是和zookeeper相連的，而對應producer和consumer的狀態保存也都是通過zookeeper完成的。對Kafka的各種操作通過其所連接的Zookeeper完成。

2.3.1 命令行客戶端

創建topic： bin/kafka-topics.sh  --create  --zookeeper  localhost:2181  --replication-factor 1  --partitions  1  --topic test 

列出所有topic： bin/kafka-topics.sh --list --zookeeper localhost:2181 

查看topic信息（包括分區、副本情況等）： kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic ，會列出分區數、副本數、副本leader節點、副本節點、活着的副本節點

往某topic生產消息： bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test 

從某topic消費消息： bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning （默認用一個線程消費指定topic的所有分區的數據）

刪除某個Kafka groupid：連接Zookeeper后用rmr命令，如刪除名為JSI的消費組： rmr /consumers/JSI 

查看消費進度：

./bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --group test-mirror-consumer-zsm --zkconnect ec2-12345.cn-north-1.compute.amazonaws.com.cn:2181/kafka/blink/0822 --topic GPS2
    各參數：
    --group指MirrorMaker消費源集群時指定的group.id
    -zkconnect指源集群的zookeeper地址
    --topic指定查的topic，沒指定則返回所有topic的消費情況

2.3.2 Java客戶端

1、Topic操作：

import kafka.admin.DeleteTopicCommand;
import kafka.admin.TopicCommand;

/**
 * @author zsm
 * @date 2016年9月27日 上午10:26:42
 * @version 1.0
 * @parameter
 * @since
 * @return
 */
public class JTopic {
    public static void createTopic(String zkAddr, String topicName, int partition, int replication) {
        String[] options = new String[] { "--create", "--zookeeper", zkAddr, "--topic", topicName, "--partitions",
                partition + "", "--replication-factor", replication + "" };
        TopicCommand.main(options);
    }

    public static void listTopic(String zkAddr) {
        String[] options = new String[] { "--list", "--zookeeper", zkAddr };
        TopicCommand.main(options);
    }

    public static void describeTopic(String zkAddr, String topicName) {
        String[] options = new String[] { "--describe", "--zookeeper", zkAddr, "--topic", topicName, };
        TopicCommand.main(options);
    }

    public static void alterTopic(String zkAddr, String topicName) {
        String[] options = new String[] { "--alter", "--zookeeper", zkAddr, "--topic", topicName, "--partitions", "5" };
        TopicCommand.main(options);
    }

    // 通過刪除zk里面對應的路徑來實現刪除topic的功能,只會刪除zk里面的信息，Kafka上真實的數據並沒有刪除
    public static void deleteTopic(String zkAddr, String topicName) {
        String[] options = new String[] { "--zookeeper", zkAddr, "--topic", topicName };
        DeleteTopicCommand.main(options);
    }

    public static void main(String[] args) {
        // TODO Auto-generated method stub

        String myTestTopic = "ZsmTestTopic";
        int myPartition = 4;
        int myreplication = 1;

        //createTopic(ConfigureAPI.KafkaProperties.ZK, myTestTopic, myPartition, myreplication);
        // listTopic(ConfigureAPI.KafkaProperties.ZK);
        describeTopic(ConfigureAPI.KafkaProperties.ZK, myTestTopic);
        // alterTopic(ConfigureAPI.KafkaProperties.ZK, myTestTopic);
        // deleteTopic(ConfigureAPI.KafkaProperties.ZK, myTestTopic);
    }

}

2、寫：（寫時可以指定key以供Kafka根據key將數據寫入某個分區，若無指定，則幾乎就是隨機找一個分區發送無key的消息，然后把這個分區號加入到緩存中以備后面直接使用——當然，Kafka本身也會清空該緩存（默認每10分鍾或每次請求topic元數據時））

package com.zsm.kfkdemo;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

import com.zsm.kfkdemo.ConfigureAPI.KafkaProperties;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

/**
 * 可以指定規則(key和分區函數)以讓消息寫到特定分區：
 * <p>
 * 1、若發送的消息沒有指定key則Kafka會隨機選擇一個分區
 * </p>
 * <p>
 * 2、否則，若指定了分區函數(通過partitioner.class)則該函數以key為參數確定寫到哪個分區
 * </p>
 * <p>
 * 3、否則，Kafka根據hash(key)%partitionNum確定寫到哪個分區
 * </p>
 * 
 * @author zsm
 * @date 2016年9月27日 上午10:26:42
 * @version 1.0
 * @parameter
 * @since
 * @return
 */
public class JProducer extends Thread {
    private Producer<String, String> producer;
    private String topic;
    private final int SLEEP = 10;
    private final int msgNum = 1000;

    public JProducer(String topic) {
        Properties props = new Properties();
        props.put("metadata.broker.list", KafkaProperties.BROKER_LIST);// 如192.168.6.127:9092,192.168.6.128:9092
        // request.required.acks
        // 0, which means that the producer never waits for an acknowledgement from the broker (the same behavior as 0.7). This option provides the lowest latency but the weakest durability guarantees
        // (some data will be lost when a server fails).
        // 1, which means that the producer gets an acknowledgement after the leader replica has received the data. This option provides better durability as the client waits until the server
        // acknowledges the request as successful (only messages that were written to the now-dead leader but not yet replicated will be lost).
        // -1, which means that the producer gets an acknowledgement after all in-sync replicas have received the data. This option provides the best durability, we guarantee that no messages will be
        // lost as long as at least one in sync replica remains.
        props.put("request.required.acks", "-1");
        // 配置value的序列化類
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        // 配置key的序列化類
        props.put("key.serializer.class", "kafka.serializer.StringEncoder");
        // 提供自定義的分區函數將消息寫到分區上，未指定的話Kafka根據hash(messageKey)%partitionNum確定寫到哪個分區
        props.put("partitioner.class", "com.zsm.kfkdemo.MyPartitioner");
        producer = new Producer<String, String>(new ProducerConfig(props));
        this.topic = topic;
    }

    @Override
    public void run() {
        boolean isBatchWriteMode = true;
        System.out.println("isBatchWriteMode: " + isBatchWriteMode);
        if (isBatchWriteMode) {
            // 批量發送
            int batchSize = 100;
            List<KeyedMessage<String, String>> msgList = new ArrayList<KeyedMessage<String, String>>(batchSize);
            for (int i = 0; i < msgNum; i++) {
                String msg = "Message_" + i;
                msgList.add(new KeyedMessage<String, String>(topic, i + "", msg));
                // msgList.add(new KeyedMessage<String, String>(topic, msg));//未指定key，Kafka會自動選擇一個分區
                if (i % batchSize == 0) {
                    producer.send(msgList);
                    System.out.println("Send->[" + msgList + "]");
                    msgList.clear();
                    try {
                        sleep(SLEEP);
                    } catch (Exception ex) {
                        ex.printStackTrace();
                    }
                }
            }
            producer.send(msgList);
        } else {
            // 單個發送
            for (int i = 0; i < msgNum; i++) {
                KeyedMessage<String, String> msg = new KeyedMessage<String, String>(topic, i + "", "Message_" + i);
                // KeyedMessage<String, String> msg = new KeyedMessage<String, String>(topic, "Message_" + i);//未指定key，Kafka會自動選擇一個分區
                producer.send(msg);
                System.out.println("Send->[" + msg + "]");
                try {
                    sleep(SLEEP);
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
        }

        System.out.println("send done");
    }

    public static void main(String[] args) {
        JProducer pro = new JProducer(KafkaProperties.TOPIC);
        pro.start();
    }
}

3、讀：（對於Consumer，需要注意 auto.commit.enable 和 auto.offset.reset 這兩個字段）

package com.zsm.kfkdemo;

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

import com.zsm.kfkdemo.ConfigureAPI.KafkaProperties;

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;

/**
 * 同一consumer group的多線程消費可以兩種方法實現：
 * <p>
 * 1、實現單線程客戶端，啟動多個去消費
 * </p>
 * <p>
 * 2、在客戶端的createMessageStreams里為topic指定大於1的線程數，再啟動多個線程處理每個stream
 * </p>
 * 
 * @author zsm
 * @date 2016年9月27日 上午10:26:42
 * @version 1.0
 * @parameter
 * @since
 * @return
 */
public class JConsumer extends Thread {

    private ConsumerConnector consumer;
    private String topic;
    private final int SLEEP = 20;

    public JConsumer(String topic) {
        consumer = Consumer.createJavaConsumerConnector(this.consumerConfig());
        this.topic = topic;
    }

    private ConsumerConfig consumerConfig() {
        Properties props = new Properties();
        props.put("zookeeper.connect", KafkaProperties.ZK);
        props.put("group.id", KafkaProperties.GROUP_ID);
        props.put("auto.commit.enable", "true");// 默認為true，讓consumer定期commit offset，zookeeper會將offset持久化，否則只在內存，若故障則再消費時會從最后一次保存的offset開始
        props.put("auto.commit.interval.ms", KafkaProperties.INTERVAL + "");// 經過INTERVAL時間提交一次offset
        props.put("auto.offset.reset", "largest");// What to do when there is no initial offset in ZooKeeper or if an offset is out of range
        props.put("zookeeper.session.timeout.ms", KafkaProperties.TIMEOUT + "");
        props.put("zookeeper.sync.time.ms", "200");
        return new ConsumerConfig(props);
    }

    @Override
    public void run() {
        Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
        topicCountMap.put(topic, new Integer(1));// 線程數
        Map<String, List<KafkaStream<byte[], byte[]>>> streams = consumer.createMessageStreams(topicCountMap);
        KafkaStream<byte[], byte[]> stream = streams.get(topic).get(0);// 若上面設了多個線程去消費，則這里需為每個stream開個線程做如下的處理

        ConsumerIterator<byte[], byte[]> it = stream.iterator();
        MessageAndMetadata<byte[], byte[]> messageAndMetaData = null;
        while (it.hasNext()) {
            messageAndMetaData = it.next();
            System.out.println(MessageFormat.format("Receive->[ message:{0} , key:{1} , partition:{2} , offset:{3} ]",
                    new String(messageAndMetaData.message()), new String(messageAndMetaData.key()),
                    messageAndMetaData.partition() + "", messageAndMetaData.offset() + ""));
            try {
                sleep(SLEEP);
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        }
    }

    public static void main(String[] args) {
        JConsumer con = new JConsumer(KafkaProperties.TOPIC);
        con.start();
    }
}

與Kafka相關的Maven依賴：

        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka_2.9.2</artifactId>
            <version>0.8.1.1</version>
            <exclusions>
                <exclusion>
                    <groupId>com.sun.jmx</groupId>
                    <artifactId>jmxri</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>com.sun.jdmk</groupId>
                    <artifactId>jmxtools</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>javax.jms</groupId>
                    <artifactId>jms</artifactId>
                </exclusion>
            </exclusions>
        </dependency>

 

3 MirrorMaker

Kafka自身提供的MirrorMaker工具用於把一個集群的數據同步到另一集群，其原理就是對源集群消費、對目標集群生產。

運行時需要指定源集群的Zookeeper地址（pull模式）或目標集群的Broker列表（push模式）。

3.1 使用

運行 ./kafka-run-class.sh kafka.tools.MirrorMaker --help 查看使用說明，如下：

Option                                  Description                            
------                                  -----------                            
--blacklist <Java regex (String)>       Blacklist of topics to mirror.         
--consumer.config <config file>         Consumer config to consume from a      
                                          source cluster. You may specify      
                                          multiple of these.                   
--help                                  Print this message.                    
--num.producers <Integer: Number of     Number of producer instances (default: 
  producers>                              1)                                   
--num.streams <Integer: Number of       Number of consumption streams.         
  threads>                                (default: 1)                         
--producer.config <config file>         Embedded producer config.              
--queue.size <Integer: Queue size in    Number of messages that are buffered   
  terms of number of messages>            between the consumer and producer    
                                          (default: 10000)                     
--whitelist <Java regex (String)>       Whitelist of topics to mirror.

3.2 啟動

./bin/kafka-run-class.sh kafka.tools.MirrorMaker --consumer.config  zsmSourceClusterConsumer.config  --num.streams 2 --producer.config zsmTargetClusterProducer.config --whitelist="ds*"
    --consumer.config所指定的文件里至少需要有zookeeper.connect、group.id兩字段
    --producer.config至少需要有metadata.broker.list字段，指定目標集群的brooker列表
    --whitelist指定要同步的topic

可以用2.3.1所說的查看消費進度來查看對原集群的同步狀況（即消費狀況）。

 

4 Kafka監控工具（KafkaOffsetMonitor）

可以借助KafkaOffsetMonitor來圖形化展示Kafka的broker節點、topic、consumer及offset等信息。

以KafkaOffsetMonitor-assembly-0.2.0.jar為例，下載后執行：

#!/bin/bash
java -Xms512M -Xmx512M -Xss1024K -XX:PermSize=256m -XX:MaxPermSize=512m -cp KafkaOffsetMonitor-assembly-0.2.0.jar \ com.quantifind.kafka.offsetapp.OffsetGetterWeb \ --zk 192.168.5.131:2181,192.168.6.132:2181,192.168.6.133:2181 \ --port 8087 \ --refresh 10.seconds \ --retain 1.days 1>./zsm-logs/stdout.log 2>./zsm-logs/stderr.log &

其中，zk按照host1:port1,host2:port2…的格式去寫即可，port為開啟web界面的端口號，refresh為刷新時間，retain為數據保留時間（單位seconds, minutes, hours, days）

 

5 Kafka集群管理工具（Kafka Manager）

kafka-manager是yahoo開源出來的項目，屬於商業級別的工具用Scala編寫。

這個管理工具可以很容易地發現分布在集群中的哪些topic分布不均勻，或者是分區在整個集群分布不均勻的的情況。它支持管理多個集群、選擇副本、副本重新分配以及創建Topic。同時，這個管理工具也是一個非常好的可以快速瀏覽這個集群的工具。

此工具以集群的方式運行，需要Zookeeper。

參考資料：http://hengyunabc.github.io/kafka-manager-install/

5.1 安裝

需要從Github下載源碼並安裝sbt工具編譯生成安裝包，生成的時間很長且不知為何一直出錯，所以這里用網友已編譯好的包 （備份鏈接）。

包為kafka-manager-1.0-SNAPSHOT.zip

>解壓：

 unzip kafka-manager-1.0-SNAPSHOT.zip 

>配置conf/application.conf里的kafka-manager.zkhosts：

 kafka-manager.zkhosts="192.168.6.131:2181,192.168.6.132:2181,192.168.6.133:2181" 

>啟動：

 ./bin/kafka-manager -Dconfig.file=conf/application.conf （啟動后在Zookeeper根目錄下可發現增加了kafka-manager目錄）

默認是9000端口，要使用其他端口可以在命令行指定http.port，此外kafka-manager.zkhosts也可以在命令行指定，如：

 ./bin/kafka-manager -Dhttp.port=9001 -Dkafka-manager.zkhosts="192.168.6.131:2181,192.168.6.132:2181,192.168.6.133:2181" 

5.2 使用

訪問web頁面，在Cluster->Add Cluster，輸入要監控的Kafka集群的Zookeeper即可。

6 進階

• 在當前的kafka版本實現中，對於zookeeper的所有操作都是由kafka controller來完成的（serially的方式）
• offset管理：kafka會記錄offset到zk中。但是，zk client api對zk的頻繁寫入是一個低效的操作。0.8.2 kafka引入了native offset storage，將offset管理從zk移出，並且可以做到水平擴展。其原理就是利用了kafka的compacted topic，offset以consumer group,topic與partion的組合作為key直接提交到compacted topic中。同時Kafka又在內存中維護了三元組來維護最新的offset信息，consumer來取最新offset信息時直接從內存拿即可。當然，kafka允許你快速checkpoint最新的offset信息到磁盤上。
• 如何確定分區數：分區數的確定與硬件、軟件、負載情況等都有關，要視具體情況而定，不過依然可以遵循一定的步驟來嘗試確定分區數：創建一個只有1個分區的topic，然后測試這個topic的producer吞吐量和consumer吞吐量。假設它們的值分別是Tp和Tc，單位是MB/s。然后假設總的目標吞吐量是Tt，那么分區數 =  Tt / max(Tp, Tc)
• Kafka Exactly Once語義與事務機制：http://www.jasongj.com/kafka/transaction/

7 參考資料

1、http://www.cnblogs.com/fanweiwei/p/3689034.html（Kafka的使用）

2、http://orchome.com/12（Broker的配置）

3、http://blog.csdn.net/lizhitao/article/details/25667831（Broker的配置）

4、http://www.jasongj.com/2015/01/02/Kafka%E6%B7%B1%E5%BA%A6%E8%A7%A3%E6%9E%90/（進階——Kafka深度解析）

5、http://www.cnblogs.com/huxi2b/p/4757098.html?utm_source=tuicool&utm_medium=referral（如何確定分區數、key、consumer線程數）