Kafka 0.10 安裝

1、下載、解壓

kafka官網(wǎng)下載0.10.x.x版本的kafka
tar -xvf 下載文件 -C /usr/local //-C 用于指定解壓目錄，不指定則為當前文件夾

2、配制

官方文檔

2.1 zookeeper配制

為簡單起見，此處使用kafka內(nèi)置的zookeeper，其配制文件為config/zookeeper.peoperties.
其主要的配制參數(shù)為，port，默認為2181

文件可以不改動，若須精細化配制，參見官檔。

2.2 kafka配制

kafka 0.10.x.x版本中，config/server.peoperties僅有三項為必須配制的。

  • broker.id
    唯一標識一個kakfa實例
  • log.dirs
    用于存儲kafka數(shù)據(jù)
  • zookeeper.connect
    配制zookeeper時最好使用zookeeper所在主機的ip地址，若使用主機名，則使用kafka客戶端的主機，必須在hosts文件是配置ip和hostname的映射。否則，應(yīng)用程序會出現(xiàn)Exception
KafKa error java.nio.channels.UnresolvedAddressException

  • listeners=PLAINTEXT://ip:9093 //此項要配制成ip +端口號，否則同樣會出現(xiàn)UnresolvedAddressException

  • 其它選項，默認即可

  • 單主機多Broker實例
    則參見 官方文檔"Step 6: Setting up a multi-broker cluster"
    僅需要增加listeners，配制項，并保證每個配制文件中broker.id，log.dir 唯一即可。

  • 多主機單Broker實例
    各個主機中的配制文件，除broker.id之外內(nèi)容可相同。

config/server-1.properties:
    broker.id=1
    listeners=PLAINTEXT://:9093
    log.dir=/tmp/kafka-logs-1
    zookeeper.connect  同上

config/server-2.properties:
    broker.id=2
    listeners=PLAINTEXT://:9094
    log.dir=/tmp/kafka-logs-2
    zookeeper.connect  同上

3、 啟動

  • 啟動zookeeper
bin/zookeeper-server-start.sh -daemon config/zookeeper.peoperties.
  • 啟動 kafka
 bin/kafka-server-start.sh -daemon config/server.peoperties.

若配制了多個broker實例，則

bin/kafka-server-start.sh -daemon config/server-1.properties 
bin/kafka-server-start.sh -daemon config/server-2.properties

4、測試

- 創(chuàng)建topic
bin/kafka-topics.sh --create --zookeeper your_host:2181 --replication-factor 1 --partitions 1 --topic topicname

- 查看是否創(chuàng)建成功
bin/kafka-topics.sh --list --zookeeper your_host:2181 topicname

- 開啟console-producer
bin/kafka-console-producer.sh --broker-list your_host:9092 --topic topicname

- 開啟 console-consumer
bin/kafka-console-consumer.sh --zookeeper your_host:2181 --topic topicname --from-beginning

在console-producer 控制臺 輸入內(nèi)容， console-consumer可以接收，則表示配制成功。若遇到如下異常：

  java.nio.BufferUnderflowException
或
  org.apache.kafka.common.protocol.types.SchemaException: Error reading field 'topics':

則說明，kafka 集群中存在不同的版本，kafka集群中每個broker是對等的，只要是注冊到同一zookeeper下的kafka就會協(xié)同。(因為是console模式所以可以排除client與server版本不同的可能)。

5、自定義 Producer與Consumer

此處使用新配制

5.1 Producer

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;
/**
 * Created by cc on 17-1-6.
 */
public class KafkaConnector {

    KafkaProducer<String, String> producer;

    public KafkaConnector(String brokerList) {
        Properties props = new Properties();

        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.CLIENT_ID_CONFIG, "producer");
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        producer = new KafkaProducer(props);
    }


    /**
     * Returns the actual number of sent records
     */
    public void sendRecord(String topic,  String key, String value) {

        try{
            ProducerRecord<String, String> record = new ProducerRecord(topic, key, value);
            producer.send(record);
        } catch (Exception e) {
            e.printStackTrace();
        }

    }


    public void close() {
        producer.close();
    }

    public static void main(String[] args) {

        String brokerlist = "192.168.4.49:9092";

        String topic = "topicname";
        KafkaConnector connector = new KafkaConnector(brokerlist);
        for(int i = 0; i < 200; i++) {
            connector.sendRecord(topic, i+"", i+"");
        }
        connector.close();
    }

}

5.2 Consumer


import java.util.*;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.TopicPartition;
import org.apache.log4j.BasicConfigurator;

/**
 * Created by cc on 17-1-6.
 */
public class Consumer {

    public KafkaConsumer<String, String> consumer;
    public Consumer(String brokerList) {
        BasicConfigurator.configure(); //防止出現(xiàn)log4j: WARN no appender警告，也可配制解決。
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest"); //"earliest" else "latest"
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "group");

        consumer = new KafkaConsumer(props);
    }


    public void process(List<String> topic) {
        consumer.subscribe(topic);
        try{
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Long.MAX_VALUE);
                for(TopicPartition partition: records.partitions()) {
                    List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
                    for(ConsumerRecord<String, String> record: partitionRecords) {
                        System.out.println(record.value());
                    }
                }
                consumer.commitSync();;

            }
        }catch (Exception e) {
            e.printStackTrace();
        } finally {
            consumer.close();
        }


    }


    public static void main(String[] args) {
        List topic = new ArrayList<String>();
        topic.add("topicname");
    
        String brokerlist = "192.168.4.49:9092";
        new Consumer(brokerlist).process(topic);

    }

}

5.3 問題解決

出現(xiàn)“l(fā)og4j:WARN No appenders could be found for logger (org.apache.kafka.clients.producer.ProducerConfig).”的解決辦法：

  • 編程解決
 BasicConfigurator.configure();  //增加此即可解決
  • 配制解決
    classpath下配制log4j.properties
最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容