KafkaOffsetMonitor是一個可以用於監控Kafka的Topic及Consumer消費狀況的工具，其配置和使用特別的方便。源專案Github地址為：https://github.com/quantifind/KafkaOffsetMonitor。
　　最簡單的使用方式是從Github上下載一個最新的KafkaOffsetMonitor-assembly-0.2.1.jar，上傳到某伺服器上，然後執行一句命令就可以執行起來。但是在使用過程中有可能會發現頁面反應緩慢或者無法顯示相應內容的情況。據說這是由於jar包中的某些js等檔案需要連線到網路，或者需要翻牆導致的。網上找的一個修改版的KafkaOffsetMonitor對應jar包，可以完全在本地執行，經過測試效果不錯。下載地址是：

一、KafkaOffsetMonitor的使用

　　因為完全沒有安裝配置的過程，所以直接從KafkaOffsetMonitor的使用開始。
　　將KafkaOffsetMonitor-assembly-0.2.0.jar上傳到伺服器後，可以新建一個指令碼用於啟動該應用。指令碼內容如下：

java -cp KafkaOffsetMonitor-assembly-0.2.0.jar \
    com.quantifind.kafka.offsetapp.OffsetGetterWeb \
    --zk m000:
 
2181,m001:2181,m002:2181 \
    --port 8088 \
    --refresh 10.seconds \
    --retain 2.days

　　各引數的作用可以參考一下Github上的描述：

• offsetStorage valid options are ”zookeeper”, ”kafka” or ”storm”. Anything else falls back to ”zookeeper”
• zk the ZooKeeper hosts
• port on what port will the app be available
• refresh how often should the app refresh and store a point in the DB
• retain how long should points be kept in the DB
• dbName where to store the history (default ‘offsetapp’)
• kafkaOffsetForceFromStart only applies to ”kafka” format. Force KafkaOffsetMonitor to scan the commit messages from start (see notes below)
• stormZKOffsetBase only applies to ”storm” format. Change the offset storage base in zookeeper, default to ”/stormconsumers” (see notes below)

• pluginsArgs additional arguments used by extensions (see below)

  　　啟動後，訪問m000:8088埠，可以看到如下頁面：
  　　
  　　在這個頁面上，可以看到當前Kafka叢集中現有的Consumer Groups。

  在上圖中有一個Visualizations選項卡，點選其中的Cluster Overview可以檢視當前Kafka叢集的Broker情況
  

  　　接下來將繼續上一篇Kafka相關的文章Kafka系列之-自定義Producer，在最後對Producer進行包裝的基礎上，分別實現一個簡單的往隨機Partition寫messge，以及自定義Partitioner的Producer，對KafkaOffsetMonitor其他頁面進行展示。

二、簡單的Producer

1、新建一個Topic

　　首先為本次試驗新建一個Topic，命令如下：

bin/kafka-topics.sh \
    --create \
    --zookeeper m000:2181 \
    --replication-factor 3 \
    --partition 3 \
    --topic kafkamonitor-simpleproducer

2、新建SimpleProducer程式碼

　　在上一篇文章中提到的Producer封裝Github程式碼的基礎上，寫了一個往kafkamonitor-simpleproducer傳送message的java程式碼。

import com.ckm.kafka.producer.impl.KafkaProducerToolImpl;
import com.ckm.kafka.producer.inter.KafkaProducerTool;

/**
 * Created by ckm on 2016/8/30.
 */
public class SimpleProducer {
    public static void main(String[] args) {
        KafkaProducerTool kafkaProducerTool = new KafkaProducerToolImpl();
        int i = 0;
        String message = "";
        while (true) {
            message = "test-simple-producer : " + i ++;
            kafkaProducerTool.publishMessage("kafkamonitor-simpleproducer", message);
        }
    }
}

　　程式執行效果：
　　

3、ConsoleConsumer消費該topic

　　用kafka自帶的ConsoleConsumer消費kafkamonitor-simpleproducer中的message。

bin/kafka-console-consumer.sh --zookeeper m000:2181 --from-beginning --topic kafkamonitor-simpleproducer

　　消費截圖如下：
　　

4、KafkaOffsetMonitor頁面

（1）在Topic List選項卡中，我們可以看到剛才新建的kafkamonitor-simpleproducer
　　
（2）點開後，能看到有一個console-consumer正在消費該topic
　　
（3）繼續進入該Consumer，可以檢視該Consumer當前的消費狀況
　　
　　這張圖片的左上角顯示了當前Topic的生產速率，右上角顯示了當前Consumer的消費速率。
　　圖片中還有三種顏色的線條，藍色的表示當前Topic中的Message數目，灰色的表示當前Consumer消費的offset位置，紅色的表示藍色灰色的差值，即當前Consumer滯後於Producer的message數目。
（4）看一眼各partition中的message消費情況
　　
　　從上圖可以看到，當前有3個Partition，每個Partition中的message數目分佈很不均勻。這裡可以與接下來的自定義Producer的情況進行一個對比。

三、自定義Partitioner的Producer

1、新建一個Topic

bin/kafka-topics.sh \
    --create \
    --zookeeper m000:2181 \
    --replication-factor 3 \
    --partition 3 \
    --topic kafkamonitor-partitionedproducer

2、Partitioner程式碼

　　邏輯很簡單，迴圈依次往各Partition中傳送message。

import kafka.producer.Partitioner;

/**
 * Created by ckm on 2016/8/30.
 */
public class TestPartitioner implements Partitioner {
    public TestPartitioner() {
    }

    @Override
    public int partition(Object key, int numPartitions) {
        int intKey = (int) key;
        return intKey % numPartitions;
    }
}

3、Producer程式碼

　　將自定義的Partitioner設定到Producer，其他呼叫過程和二中類似。

import com.ckm.kafka.producer.impl.KafkaProducerToolImpl;
import com.ckm.kafka.producer.inter.KafkaProducerTool;

/**
 * Created by ckm on 2016/8/30.
 */
public class PartitionedProducer {
    public static void main(String[] args) {
        KafkaProducerTool kafkaProducerTool = new KafkaProducerToolImpl();
        kafkaProducerTool.getProducerProperties().put("partitioner.class", "TestPartitioner");
        int i = 0;
        String message = "";
        while (true) {
            message = "test-partitioner-producer : " + i;
            System.out.println(message);
            kafkaProducerTool.publishPartitionedMessage("kafkamonitor-partitionedproducer", i + "", message);
            i ++;
        }
    }
}

　　程式碼執行效果如下圖：
　　

4、ConsoleConsumer消費Message

bin/kafka-console-consumer.sh --zookeeper m000:2181 --from-beginning --topic kafkamonitor-partitionedproducer

　　消費效果如下圖：
　　

5、KafkaOffsetMonitor頁面

　　其他頁面與上面的類似，這裡只觀察一下每個partition中的message數目與第二節中的對比。可以看到這裡每個Partition中message分別是很均勻的。
　　

注意事項：
　　注意這裡有一個坑，預設情況下Producer往一個不存在的Topic傳送message時會自動建立這個Topic。由於在這個封裝中，有同時傳遞message和topic的情況，如果呼叫方法時傳入的引數反了，將會在Kafka叢集中自動建立Topic。在正常情況下，應該是先把Topic根據需要建立好，然後Producer往該Topic傳送Message，最好把Kafka這個預設自動建立Topic的功能關掉。
　　那麼，假設真的不小心建立了多餘的Topic，在刪除時，會出現“marked for deletion”提示，只是將該topic標記為刪除，使用list命令仍然能看到。如果需要調整這兩個功能的話，在server.properties中配置如下兩個引數：