最近在搞spark streaming，很自然的前端对接的就是kafka。不过在kafka的使用中还是遇到一些问题，比如mirrormaker莫名其妙的丢失数据[原因稍后再说]，消费数据offset错乱[之后介绍spark streaming的时候再解释]

总之，还是遇到了不少的问题。本篇就从下面几个方面介绍一下kafka：

1. 基本介绍
2. 安装与helloworld
3. producer
4. consumer
5. mirror maker跨集群同步
6. 控制台

基本介绍

Kafka是一款分布式的消息队列框架，它由三个重要的部分组成：

1. Producer 消息的生产者，负责生产消息
2. Broker 消息的存储，负责消息的持久化与高可用
3. Consumer 消息的消费者，负责消费消息

大致的结构如下：

消息则是通过topic进行标识，每个topic可以有多个partition分区组成。每一个parition内部消息是按照顺序写入的，所有的partition加起来才是全部的数据，也就是说kafka并不能保证全局有序，只能保证在某一个partition内部是有序的。

消费者消费数据的时候是根据一个叫做offset的游标来记录消费的位置，可以通俗的把它理解成递增的id。

消费者可以由多个组成一个消费者组，同一个消费者组内的数据不会重复消费。不过消费者的数量跟partition的数量是有关系的，如果只有一个partition，那么即便是由10个消费者，同一时间也只能由一个消费者进行消费。

另外，broker是负责消息的持久化，前面提到过消息是通过partition组织在一起的，物理上则是通过一个log文件来记录。如果有一条消息写入，就会追加到log文件的末尾，当大小超过一定的阈值后，就新建一个log文件。如果log文件的修改时间超过一定的阈值，kafka还会清理掉该文件。

原理的东西就简单说这么多，下面来看看安装与体验吧！

安装与hello world

按照官方文档的步骤，是最快的入门方式：

下载安装包

去官方下载地址下载安装包，并参照对应的版本的文档即可，下载后执行下面的命令：

> tar -xzf kafka_2.11-0.9.0.0.tgz
> cd kafka_2.11-0.9.0.0

启动zookeeper

如果方便的话，最好自己额外安装zookeeper，或者与其他的组建公用一个zk，否则单独为了kafka运行一个zk还是挺浪费资源的。

> bin/zookeeper-server-start.sh config/zookeeper.properties

最好不要随意修改zk的地址，2181是默认的端口号，如果修改，后面启动kafka会很麻烦，修改的地方会很多。

启动kafka-broker

bin/kafka-server-start.sh config/server.properties

创建主题并查看

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

启动producer

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

启动consumer

bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning

至此，单机版的kafka就搭建完成了！如果要创建kafka的集群，可以直接

producer例子

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

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

public class KafkaProducer {
    private static final String TOPIC = "test"; //kafka创建的topic
    private static final String CONTENT = "This is a single message"; //要发送的内容
    private static final String BROKER_LIST = "xxxx:9092"; //broker的地址和端口
    private static final String SERIALIZER_CLASS = "kafka.serializer.StringEncoder"; // 序列化类

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("serializer.class", SERIALIZER_CLASS);
        props.put("metadata.broker.list", BROKER_LIST);


        ProducerConfig config = new ProducerConfig(props);
        Producer<String, String> producer = new Producer<String, String>(config);

        //Send one message.
        KeyedMessage<String, String> message =
                new KeyedMessage<String, String>(TOPIC, CONTENT);
        producer.send(message);

        //Send multiple messages.
        List<KeyedMessage<String,String>> messages =
                new ArrayList<KeyedMessage<String, String>>();
        for (int i = 0; i < 100; i++) {
            messages.add(new KeyedMessage<String, String>
                    (TOPIC, i+"Multiple message at a time. " + i));
        }
        producer.send(messages);
        producer.close();
    }
}

执行后，如果有一个consumer启动，就可以看到消息输出。

consumer例子

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

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

public class KafkaConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("zookeeper.connect", "xxxx:2181");
        props.put("group.id", "t1");

        Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
        topicCountMap.put("xxx-topic", 1);
        ConsumerConnector consumer = Consumer.createJavaConsumerConnector(new ConsumerConfig(props));
        Map<String, List<KafkaStream<byte[], byte[]>>> msgStreams = consumer.createMessageStreams(topicCountMap);
        List<KafkaStream<byte[], byte[]>> msgStreamList = msgStreams.get("test");

        for(KafkaStream stream : msgStreamList){
            ConsumerIterator<byte[], byte[]> iterator = stream.iterator();
            while(iterator.hasNext()) {
                String message = new String(iterator.next().message());
                if(message.contains("xxxx")){
                    System.out.println(message);
                }
            }
        }
    }
}

跨集群同步——mirror maker

如果公司有云环境，可能还涉及到多个集群环境数据的同步。那么官方提供了一个mirrormaker的工具，它其实就是封装了一个consumer和一个producer，把一个集群的数据，直接消费到另一个集群。

代码可以参考github：
https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/tools/MirrorMaker.scala

文档可以参考：
https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=27846330

我这里介绍一下它的用法，首先启动的脚本，官方已经封装到kafka解压后的bin目录下。

主要用到了kafka-run-class.sh，kafka-mirror-maker.sh脚本其实就是对它的一层封装：

exec $(dirname $0)/kafka-run-class.sh kafka.tools.MirrorMaker $@

然后需要创建两个配置文件，分别是consumer的配置文件和producer的配置文件：

consumer.properties

zookeeper.connect=xxxx:2181 
group.id=test-mirror

zookeeper.connect是想要消费的集群的zk地址，group.id是消费者组的id，一定别跟其他的mirrormaker搞到一起哈！[这就是我开篇遇到的问题原因]。

producer.properties

zk.connect=localhost:2181
bootstrap.servers=localhost:9092

zk.connect是消息即将存储的zk地址, bootstrap.servers是消息即将存储的broker地址。（我试过没有bootstrap.servers的话，会报错）

然后执行下面的命令，启动脚本即可：

./kafka-run-class.sh kafka.tools.MirrorMaker --consumerrties --producer.config producer.properties --whitelist test --num.streams 2

num.streams控制了消费者的个数，必须要设置的。

这样就开启了mirrormaker服务，可以看到第一个集群的所有消息，都同步到了第二个集群。

控制台主要功能介绍

控制台可以安装kafka-manager进行监控与管理，安装的教程可以参考：
http://blog.csdn.net/lsshlsw/article/details/47300145

集群概况

主题

broker

消费者

参考

• 官方文档