centos7單機安裝kafka，進行生產者消費者測試

原文出處：http://www.yund.tech/zdetail.html?type=1&id=3028469704c7976aef5b824811dd3bf5  

作者：jstarseven 

---

  

一、kafka介紹

Kafka是最初由Linkedin公司開發，是一個分布式、分區的、多副本的、多訂閱者，基於zookeeper協調的分布式日志系統（也可以當做MQ系統），常見可以用於web/nginx日志、訪問日志，消息服務等等，Linkedin於2010年貢獻給了Apache基金會並成為頂級開源項目。

主要應用場景是：日志收集系統和消息系統。

Kafka主要設計目標如下：

• 以時間復雜度為O(1)的方式提供消息持久化能力，即使對TB級以上數據也能保證常數時間的訪問性能。
• 高吞吐率。即使在非常廉價的商用機器上也能做到單機支持每秒100K條消息的傳輸。
• 支持Kafka Server間的消息分區，及分布式消費，同時保證每個partition內的消息順序傳輸。
• 同時支持離線數據處理和實時數據處理。
• 支持在線水平擴展

二、kafka架構圖

三、kafka安裝與測試

1、配置JDK環境
Kafka 使用Zookeeper 來保存相關配置信息，Kafka及Zookeeper 依賴Java 運行環境，從oracle網站下載JDK 安裝包，解壓安裝

tar zxvf jdk-8u171-linux-x64.tar.gz
mv jdk1.8.0_171 /usr/local/java/

設置Java 環境變量：

#java 
export JAVA_HOME=/usr/local/java/jdk1.8.0_171
export PATH=$PATH:$JAVA_HOME/bin
export CLASSPATH=.:$JAVA_HOME/lib/tools.jar:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib:$JAVA_HOME/jre/lib

2、安裝kafka

下載地址：http://kafka.apache.org/downloads

cd /opt
wget http://mirror.bit.edu.cn/apache/kafka/2.3.0/kafka_2.11-2.3.0.tgz
tar zxvf kafka_2.11-2.3.0.tgz
mv kafka_2.11-2.3.0 /usr/local/apps/
cd /usr/local/apps/
ln -s kafka_2.11-2.3.0 kafka

3、啟動測試

（1）啟動Zookeeper服務

cd /usr/local/apps/kafka
#執行腳本
bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
#查看進程
jps

（2）啟動單機Kafka服務

#執行腳本
bin/kafka-server-start.sh config/server.properties
#查看進程
jps

（3）創建topic進行測試

#執行腳本
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

（4）查看topic列表

#執行腳本
bin/kafka-topics.sh --list --zookeeper localhost:2181
輸出：test

（5）生產者消息測試

#執行腳本(使用kafka-console-producer.sh 發送消息)
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

（6）消費者消息測試

#執行腳本(使用kafka-console-consumer.sh 接收消息並在終端打印)
bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning

 4、單機多broker集群配置

　單機部署多個broker，不同的broker，設置不同的id、監聽端口、日志目錄

    cp config/server.properties config/server-1.properties 
    vim server-1.properties
    #修改：
    broker.id=1
    port=9093
    log.dir=/tmp/kafka-logs-1
    #啟動Kafka服務
    bin/kafka-server-start.sh config/server-1.properties &

 5、java代碼實現生產者消費者

　　（1）maven項目添加kafka依賴

<dependency>
         <groupId>org.apache.kafka</groupId>
         <artifactId>kafka-clients</artifactId>
         <version>2.3.0</version>
</dependency>

　　（2）java代碼實現

package com.server.kafka;

import org.apache.kafka.clients.consumer.ConsumerConfig;
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.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Collections;
import java.util.Properties;
import java.util.Random;


public class KafakaExecutor {

    public static String topic = "test";

    public static void main(String[] args) {
        new Thread(()-> new Producer().execute()).start();
        new Thread(()-> new Consumer().execute()).start();
    }

    public static class Consumer {

        private void execute() {
            Properties p = new Properties();
            p.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.21.181:9092");
            p.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
            p.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
            p.put(ConsumerConfig.GROUP_ID_CONFIG, topic);

            KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(p);
            // 訂閱消息
            kafkaConsumer.subscribe(Collections.singletonList(topic));

            while (true) {
                ConsumerRecords<String, String> records = kafkaConsumer.poll(100);
                for (ConsumerRecord<String, String> record : records) {
                    System.out.println(String.format("topic:%s,offset:%d,消息:%s", //
                            record.topic(), record.offset(), record.value()));
                }
            }
        }
    }


    public static class Producer {

        private void execute() {
            Properties p = new Properties();
            //kafka地址，多個地址用逗號分割
            p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.21.181:9092");
            p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
            p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
            KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(p);

            try {
                while (true) {
                    String msg = "Hello," + new Random().nextInt(100);
                    ProducerRecord<String, String> record = new ProducerRecord<>(topic, msg);
                    kafkaProducer.send(record);
                    System.out.println("消息發送成功:" + msg);
                    Thread.sleep(500);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                kafkaProducer.close();
            }
        }

    }
}

　　（3）測試結果（上面使用腳本命令執行消費者的終端也會同步輸出消息數據）

 參考：https://www.cnblogs.com/frankdeng/p/9310684.html 

---

 -END-