Kafka 學(xué)習(xí)筆記

內(nèi)容大部分引用自Info - Apache Kafka：下一代分布式消息系統(tǒng)

原文作者Abhishek Sharma的項目Kafka-Message-Server

介紹

Kafka是使用scala語言開發(fā)镜盯，類似于RabbitMQ的分布式消息系統(tǒng)。
Kafka是分布式的，它通過可以多個broker組成一個集群骡显。
Kafka依賴于Zookeeper。

概念

Topic(話題) 特定類型的消息流奸鬓。消息是字節(jié)的有效負(fù)載(Payload)撇眯，話題是消息的分類或種子(Feed)名。

Producer(生產(chǎn)者) 能夠發(fā)布消息到話題的任何對象旷祸。

Broker(代理) 或稱Kafka集群。用于保存消息的服務(wù)器讼昆。

Consumer(消費(fèi)者) 可以訂閱一個或多個話題托享，并從Broker拉取數(shù)據(jù)，從而消費(fèi)這些已發(fā)布的消息。

Kafka生產(chǎn)者, 消費(fèi)者 & 代理

架構(gòu)

生產(chǎn)者使用自己的序列化方法對消息內(nèi)容進(jìn)行編碼闰围。然后向broker發(fā)起消息赃绊。為了提高效率，一個發(fā)布請求中可以包含一組消息羡榴。

消費(fèi)者訂閱話題碧查，并為話題創(chuàng)建一個或多個消息流。發(fā)布到該話題的消息被均衡的分發(fā)到這些流中校仑。

每個消息流為不斷產(chǎn)生的消息提供了迭代接口忠售。

消費(fèi)者迭代流中每一條消息，并處理消息的有效負(fù)載迄沫。

迭代器不會停止稻扬。如果當(dāng)前沒有消息，迭代器將阻塞直至有新的消息發(fā)布到該話題羊瘩。

Kafka支持點(diǎn)到點(diǎn)分發(fā)模型(Proint-to-point delivery model)泰佳，即多個消費(fèi)者共同消費(fèi)隊列中某個消息的單個副本；也支持發(fā)布-訂閱模型(Publish-subscribe model)尘吗，即多個消費(fèi)者接收自己的消息副本逝她。

kafka-proc-topic-part-con.png

存儲

kafka的存儲，話題的每個分區(qū)對應(yīng)一個邏輯日志摇予。物理上汽绢，一個日志為相同大小的一段分組文件。

每次生產(chǎn)者發(fā)布消息到一個分區(qū)侧戴，代理就將消息追加到最后一個段文件中宁昭。

當(dāng)發(fā)布的消息數(shù)量達(dá)到設(shè)定值或經(jīng)過一段時間后，段文件真正寫入磁盤中酗宋。

寫入完成后积仗，消息公開給消費(fèi)者。

與傳統(tǒng)的消息不同蜕猫，kafka系統(tǒng)中存儲的消息沒有明確的id寂曹，而是通過日志的邏輯偏移量來公開。相比其他方式回右，這種處理更為高效隆圆。

消費(fèi)者始終從特殊分區(qū)順序的獲取消息。

代理

不同于其他消息系統(tǒng)翔烁，kafka代理是無狀態(tài)的渺氧，即消費(fèi)者必須維護(hù)已消費(fèi)的狀態(tài)消息，而代理完全不管蹬屹。

這種設(shè)計的創(chuàng)新在于:

• 代理以一個基于時間的SLA應(yīng)用于保留策略侣背。當(dāng)消息在代理中超過一定時間后白华，將會被自動刪除。

• 消費(fèi)者可以故意倒回到老的偏移量再次消費(fèi)數(shù)據(jù)贩耐。雖然這違法了隊列的常見約定弧腥，但常見于許多業(yè)務(wù)中。

與zookeeper的關(guān)系

kafka使用ZooKeeper用于管理潮太、協(xié)調(diào)代理管搪。每個Kafka代理通過Zookeeper協(xié)調(diào)其他Kafka代理。

當(dāng)Kafka系統(tǒng)中新增了代理或某個代理失效時消别，Zookeeper服務(wù)將通知生產(chǎn)者和消費(fèi)者抛蚤。

生產(chǎn)者與消費(fèi)者據(jù)此開始與其他代理協(xié)調(diào)工作。

安裝

wget http://mirrors.cnnic.cn/apache/kafka/0.8.2.1/kafka_2.11-0.8.2.1.tgz
tar -xzvf kafka_2.11-0.8.2.1.tgz

配置

config/server.properties

# 如果配置多個kafka節(jié)點(diǎn)寻狂，id需設(shè)置為不同的值
broker.id=1

# !!務(wù)必將host.name配置為ip地址。
# 在java代碼里連接kafka時朋沮，服務(wù)端會把host.name的值傳給zookeeper
# 如果使用默認(rèn)配置的localhost蛇券，會出現(xiàn)連接失敗的異常
host.name=192.168.1.1

port=9092

log.dir=./logs

# 如果有多個zookeeper服務(wù)，用,號隔開即可樊拓。
# zookeeper使用默認(rèn)配置的2181端口
zookeeper.connect=192.168.1.1:2181
zookeeper.connection.timeout.ms=6000

啟動

bin/

# 先啟動zookeeper服務(wù)
./zookeeper-server-start.sh ../config/zookeeper.properties &
# 再啟動kafka服務(wù)
./kafka-server-start.sh ../config/server.properties

測試

bin/

創(chuàng)建一個Topic

./kafka-topics.sh --create --zookeeper 192.168.1.1:2181 --replication-factor 2 --partitions 1 --topic test

查看Topic信息

./kafka-topics.sh --describe --zookeeper 192.168.1.1:2181 --topic test

啟動生產(chǎn)者(producer)生產(chǎn)該Topic的消息

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

啟動消費(fèi)者(consumer)消費(fèi)該Topic的消息

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

Java代碼

Producer 測試

import java.util.Date;
import java.util.Properties;

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

public class KafkaProducer {
    
    public static void testProducer() {
        Properties props = new Properties();
        props.put("metadata.broker.list", "192.168.1.1:9092");
        props.put("serializer.class", StringEncoder.class.getName());
        //props.put("partitioner.class", );
        props.put("request.required.arks", "1");
        
        ProducerConfig config = new ProducerConfig(props);
        
        Producer<String, String> producer = new Producer<String, String>(config);
        
        String msg = new Date() + " - hello world : 測試 " ;
        KeyedMessage<String, String> data = new KeyedMessage<String, String>("test", msg);
        producer.send(data);
        producer.close();
        System.out.println("--> producer sended： " + msg);
    }
    
    public static void main(String[] args) {
        testProducer();
    }
}

Consumer 測試

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.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;

public class KafkaConsumer {

    private final ConsumerConnector consumer;
    private final String topic;
    private ExecutorService executor;

    public KafkaConsumer(String a_zookeeper, String a_groupId, String a_topic) {
        this.consumer = kafka.consumer.Consumer
                .createJavaConsumerConnector(createConsumerConfig(a_zookeeper,
                        a_groupId));

        this.topic = a_topic;
    }

    private static ConsumerConfig createConsumerConfig(String a_zookeeper,
            String a_groupId) {
        Properties props = new Properties();
        props.put("zookeeper.connect", a_zookeeper);
        props.put("group.id", a_groupId);
        props.put("zookeeper.session.timeout.ms", "1000");
        props.put("zookeeper.sync.time.ms", "1000");
        props.put("auto.commit.interval.ms", "1000");
        props.put("auto.offset.reset", "smallest");

        return new ConsumerConfig(props);
    }

    public void shutdown() {
        if (consumer != null)
            consumer.shutdown();
        if (executor != null)
            executor.shutdown();
    }

    public void run(int a_numThreads) {
        Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
        topicCountMap.put(topic, new Integer(a_numThreads));
        Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer
                .createMessageStreams(topicCountMap);
        List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic);

        System.out.println("streams.size = " + streams.size());

        // now launch all the threads
        //
        executor = Executors.newFixedThreadPool(a_numThreads);

        // now create an object to consume the messages
        //
        int threadNumber = 0;
        for (final KafkaStream stream : streams) {
            executor.submit(new ConsumerTest(stream, threadNumber));
            threadNumber++;
        }
    }

    public static void main(String[] args) {

        String zooKeeper = "192.168.212.100:2181";
        String groupId = "group1";
        String topic = "test";

        int threads = 3;

        KafkaConsumer example = new KafkaConsumer(zooKeeper, groupId, topic);

        example.run(threads);

    }
    
    public class ConsumerTest implements Runnable {

        private KafkaStream m_stream;
        private int m_threadNumber;

        public ConsumerTest(KafkaStream a_stream, int a_threadNumber) {
            m_threadNumber = a_threadNumber;
            m_stream = a_stream;
        }

        public void run() {
            System.out.println("calling ConsumerTest.run()");
            ConsumerIterator<byte[], byte[]> it = m_stream.iterator();

            while (it.hasNext()) {
                System.out.println("--> consumer  Thread " + m_threadNumber + ": "
                        + new String(it.next().message()));
            }

            System.out.println("Shutting down Thread: " + m_threadNumber);
        }
    }

}