初步認識Kafka

時間 2019-11-21

標籤初步認識 kafka 欄目 Kafka 简体版

原文原文鏈接

1.什麼是Kafkajava

Apache Kafka是一個開源的分佈式消息隊列 (生產者消費者模式) Apache Kafka 目標:構建企業中統一的. 高通量的,低延時的消息平臺 ; 大多的消息隊列是基於JMS標準實現的Apache Kafka l相似於JMS的實現.node

2.Kafka的特色算法

做爲緩衝(流量消減),來異構,解耦系統apache

3.基本架構bootstrap

Kafka Cluster：由多個服務器組成。每一個服務器單獨的名字broker（掮客）。緩存

kafka broker：kafka集羣中包含的服務器服務器

Kafka Producer：消息生產者、發佈消息到 kafka 集羣的終端或服務。架構

Kafka consumer：消息消費者、負責消費數據。負載均衡

Kafka Topic: 主題，一類消息的名稱。存儲數據時將一類數據存放在某個topic下，消費數據也是消費一類數據。異步

訂單系統：建立一個topic，叫作order。

用戶系統：建立一個topic，叫作user。

商品系統：建立一個topic，叫作product。

注意：Kafka的元數據都是存放在zookeeper中。

4.Kafka的基本使用

4.1使用腳本操做Kafka

1) 建立一個topic

./kafka-topics.sh --create --zookeeper node01:2181 --replication-factor 1 --partitions 1 --topic order

2) 使用Kafka 自帶一個命令客戶端啓動一個生產者,生產數據
./kafka-console-producer.sh --broker-list node01:9092 --topic order

3) 使用Kafka自帶一個命令客戶端啓動一個消費者,消費數據

./kafka-console-consumer.sh --bootstrap-server node01:9092 --topic order
該消費語句，只能獲取最新的數據，要想歷史數據，須要添加選項-

4)查看有哪些topic

./kafka-topics.sh --list --zookeeper node01:2181

5) 查看某一個具體的Topic的詳細信息

./kafka-topics.sh --describe --topic order --zookeeper node01:2181

6) 刪除topic

./kafka-topics.sh --delete --topic order --zookeeper node01:2181

4.2 使用java API 操做kafka

第一步: 添加kafka相關的依賴

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

4.2.1 編寫生產者 :

// kakfa的生產者
public class KafkaProducerTest {

public static void main(String[] args) {

//1. 建立 kafka的生產者的對象

Properties props = new Properties();

props.put("bootstrap.servers", "192.168.72.141:9092,192.168.72.142:9092,192.168.72.143:9092");
props.put("acks", "all"); // 消息確認機制 : all 最高級別, 保證數據不會丟失
props.put("retries", 0); // 重試 : 0 表示發送失敗, 不會重試
props.put("batch.size", 16384); // 發送數據時候一批數據的大小默認值: 16384字節(16)
props.put("linger.ms", 1); // 每次發送數據間隔時間
props.put("buffer.memory", 33554432); // 緩存池的大小: 默認值: 33554432 32M
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // 效率, java的序列化慢
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

KafkaProducer<String,String> producer = new KafkaProducer<String, String>(props);
//2. 發送消息
ProducerRecord<String, String> record = new ProducerRecord<String, String>("order", "我是JavaAPI發過來...");
producer.send(record);
//3. 關閉資源

producer.close();
}
}

4.2.2 編寫消費者

package com.itheima.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Properties;

// kafka的消費者
public class KafkaConsumerTest {

public static void main(String[] args) {

Properties props = new Properties();
props.put("bootstrap.servers", "192.168.72.141:9092,192.168.72.142:9092,192.168.72.143:9092");
props.put("group.id", "test"); // 組id號
props.put("enable.auto.commit", "true"); // 開啓自動提交
props.put("auto.commit.interval.ms", "1000"); // 每隔多長時間自動提交一次
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
//1. 建立 kafka的消費者的對象
KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);

//2. 讓消費者訂閱一個topic
consumer.subscribe(Arrays.asList("order"));

//3. 獲取消息
while (true) {
ConsumerRecords<String, String> records = consumer.poll(100); // 取出元素,
for (ConsumerRecord<String, String> record : records) {
System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());

}
}

5 Apache Kafka原理