kafka 上手指南：單節點

時間  2019-11-08
標籤 kafka 上手 指南 節點 欄目 Kafka 简体版
原文   原文鏈接

你們好，我叫謝偉，是一名程序員。python

今天的主題：kafka 使用指南，單節點版本。nginx

1. 使用場景

若是你是一名後端工程師，設計的應用正常的線上運行，某次秒殺活動，忽然間把系統搞崩了，排查系統發現不少的流量沒有處理，致使系統掛了，這個時候有兩種思路： 1. nginx 反向代理，把更多的請求轉發給內部網絡的服務器上進行處理，達到一個負載均衡的目的 2. 使用消息系統，將更多的請求使用中間件「緩存」起來，再從這個系統中不斷的取到緩存的請求，進行進一步的處理。git

後者使用到的消息系統，就是kafka 的一個使用場景。程序員

那麼什麼是 kafka?github

kafka 是一個分佈式消息系統，目前已定位爲分佈式流式處理平臺。golang

簡單的說一個系統A 將消息發給消息系統，一個系統B 再從消息系統中取到消息，進行後續的處理。算法

常見的用來描述 kafka 應用場景的一個詞是：削峯填谷，削減波峯流量，填充波谷流量，使系統儘可能的平滑。docker

由此得處：kafka 的三個典型應用場景json

  • 消息系統
  • 存儲系統
  • 分佈式流式處理平臺

消息系統是目前最普遍的應用；消息傳輸須要存儲起來，供後續系統拉取，故也能夠看成存儲系統；拉取消息以後，其實也是供後續系統處理，那麼爲何不把數據處理也包含再kafka 系統中？分佈式流式處理平臺，大概就是這個意思。後端

下文陳述最核心的應用：消息系統

2. 基本概念

一條消息由系統A 產生，發往消息系統，系統B 從消息系統中拉取，這其中涉及到不少的概念。

  • 系統A 稱爲生產者 producer，目的是發送消息
  • 消息系統稱爲 broker，本質是服務進程目的是接受生產者的消息、消費者的消息拉取請求、持久化
  • 系統B 稱爲消費者 consumer, 目的是拉取消息系統中的消息

針對生產者、消費者有不一樣的設置參數，決定了生產者、消費者的不一樣行爲。

生產者要發送消息，首先要知道發往何處，即要知道 broker 的地址，知道 broker 的地址，broker（kafka server） 的設置約束了持久化存儲的地址及其餘行爲，除此以外，如何區分發的消息的類型不一樣呢？kafka 系統給這個區分消息的概念取了個邏輯概念：Topic ， 即生產者指定的 Topic 不一樣，存儲的地址就不一樣。

針對 Topic，簡單的場景是，不斷的往裏面發內容，持久化存儲就不斷以追加的模式存儲，簡單場景沒什麼問題，問題是消息數據過多的話，不利於系統消費，很簡單的想法，分不一樣的「文件」追加存儲，把總體規模縮小，這個概念在 kafka 中稱之爲分區：partition. 消息能夠不斷的以追加的模式不斷的發往分區內，分區有編號，起始位 0 ，消息追加模式存儲在分區內，會給一個編號 offset

消費者從 broker 系統中拉取消息，首先要知道broker 地址，其次須要知道 Topic，更細化的還能夠設置哪一個分區，哪一個偏移量 offset 開始，消費消息。

那消息萬一丟了咋整？一個簡單的作法就是冗餘備份：Replication，多份備份，其中有一個是 Leader , 其餘的是 follower， leader 的做用是和消息對接，follower 不直接和消息對接，只負責和 leader 對接，不斷的同步數據。

多個 broker 構成 kafka 集羣，萬一一個掛了 kafka 系統依靠 zookeeper 進行從新選舉產生新leader。

kafka cluster:

image

kafka topic: 分區概念

image

kafka 集羣：

image

3. 客戶端使用

基於上述概念：那麼如何構建一個Kafka 服務，完成消息系統呢？

  • 啓動服務進程：broker

僞代碼：

type Broker struct{
    Addr 
    Config
    ...
}
複製代碼
  • 生產者鏈接 broker

僞代碼：

type Producer struct{
    Config
    Message 
    ...
}

複製代碼
  • 消費者鏈接 broker

僞代碼

type Consumer strcut{
    Config
    Topic 
    Partitions
    Offset
    ...
}
複製代碼

基本的思路：

  • 啓動kafka服務
  • 系統A 鏈接服務，發送消息
  • 系統B 鏈接服務，消費消息

結合官網的示例：如何完成最基本的消息收發。

下載安裝包：kafka_2.12-2.3.0.tgz

  • 2.12 指編譯器版本
  • 2.3.0 指kafka 版本

解壓以後，最重要的有兩目錄：

  • bin : 一系列的腳本，好比啓動 zookeeper 服務，建立 topic，生產者生產消息，消費者消費消息等
zookeeper-server-start.sh
zookeeper-server-stop.sh
kafka-configs.sh
kafka-console-consumer.sh
kafka-console-producer.sh
kafka-consumer-groups.sh
kafka-topics.sh
kafka-server-start.sh
kafka-server-stop.sh
...

複製代碼
  • config: 配置文件：好比配置 zookeeper 端口，配置kafka 日誌存儲目錄、對外端口，消息最大容量，保存時常等
zookeeper.properties
server.properties
producer.properties
consumer.properties
...
複製代碼

大概200多個參數吧，很差意思，我記不住。那怎麼辦？不學了嗎，那掙不了錢，漲不了工資啊。

基本默認設置，部分按分類設置：

  • zookeeper.properties

kafka 依賴於 zookeeper 分佈式協調

dataDir=/tmp/zookeeper
clientPort=2181
複製代碼

記住這個默認的 clientPort=2181

  • server.properties

kafka server 服務

log.dirs=/tmp/kafka-logs //日誌存儲目錄
log.retention.hours=168 // 日誌存儲時長
broker.id=0 // 默認 broker id，集羣方式的 kafka 設置，給每一個 broker 編號
listeners=PLAINTEXT://:9092 // 對外提供的服務入口地址
zookeeper.connect=localhost:2181 // ZooKeeper集羣地址
...
複製代碼
  • producer.properties

約定消息等的內容

  • consumer.properties

約定消費消息等的內容

配置好配置參數後：

  • 啓動 zookeeper
> bin/zookeeper-server-start.sh config/zookeeper.properties
複製代碼
  • 啓動 kafka 服務進程
> bin/kafka-server-start.sh config/server.properties
複製代碼

建立topic, 查詢 topic 等可使用：kafka-topics.sh

生產者生產消息可使用：kafka-console-producer.sh

消費者消費消息可使用：kafka-console-consumer.sh

固然，這些操做，通常只供測試使用，實際的使用是使用對應變成語言的客戶端。

4. 演示

kafka go版本客戶端：

下載安裝：

go get -u -v github.com/Shopify/sarama
複製代碼

4.1 生產者

系統 A

  • 生產者
type KafkaAction struct {
	DataSyncProducer  sarama.SyncProducer
	DataAsyncProducer sarama.AsyncProducer
}
複製代碼
// 同步方式

func newDataSyncProducer(brokerList []string) sarama.SyncProducer {
	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll // Wait for all in-sync replicas to ack the message
	config.Producer.Retry.Max = 5                    // Retry up to 10 times to produce the message
	config.Producer.Return.Successes = true
	config.Producer.Partitioner = sarama.NewRoundRobinPartitioner
	producer, err := sarama.NewSyncProducer(brokerList, config)
	if err != nil {
		log.Fatalln("Failed to start Sarama producer1:", err)
	}
	return producer

}

複製代碼
// 異步方式
func newDataAsyncProducer(brokerList []string) sarama.AsyncProducer {
	config := sarama.NewConfig()
	sarama.Logger = log.New(os.Stdout, "[KAFKA] ", log.LstdFlags)
	config.Producer.RequiredAcks = sarama.WaitForLocal       // Only wait for the leader to ack
	config.Producer.Compression = sarama.CompressionSnappy   // Compress messages
	config.Producer.Flush.Frequency = 500 * time.Millisecond // Flush batches every 500ms
	config.Producer.Partitioner = sarama.NewRoundRobinPartitioner
	producer, err := sarama.NewAsyncProducer(brokerList, config)
	if err != nil {
		log.Fatalln("Failed to start Sarama producer2:", err)
	}
	go func() {
		for err := range producer.Errors() {
			log.Println("Failed to write access log entry:", err)
		}
	}()
	return producer
}

複製代碼

還記得生產者有一系列配置參數嗎？config 就這這個做用，有默認值，能夠本身設置對應的值。

好比：壓縮算法

config.Producer.Compression = sarama.CompressionSnappy
複製代碼

經常使用的壓縮算法有：

  • gzip
  • snappy
  • lz4
  • zstd

不一樣的壓縮算法主要在壓縮比和吞吐量不一樣。

好比分區規則

config.Producer.Partitioner = sarama.NewRoundRobinPartitioner
複製代碼

經常使用的分區規則：

  • 輪詢機制
  • 隨機分區
  • 按 key 分區

好比：發送消息是否返回成功與否

onfig.Producer.RequiredAcks = sarama.WaitForLocal
複製代碼
  • 消息：生產者只傳遞字節組數據。

接口

type Encoder interface {
	Encode() ([]byte, error)
	Length() int
}
複製代碼

發送的消息須要實現Encoder 接口，即定義的消息結構體須要實現 Encode 和 Length 方法。

type SendMessage struct {
	Method  string `json:"method"`
	URL     string `json:"url"`
	Value   string `json:"value"`
	Date    string `json:"date"`
	encoded []byte
	err     error
}

func (S *SendMessage) Length() int {
	b, e := json.Marshal(S)
	S.encoded = b
	S.err = e
	return len(string(b))
}
func (S *SendMessage) Encode() ([]byte, error) {
	return S.encoded, S.err
}
複製代碼
  • 發送消息
func (K *KafkaAction) Do(v interface{}) {
	message := v.(SendMessage)
    // 發送的消息返回分區和偏移量
	partition, offset, err := K.DataSyncProducer.SendMessage(&sarama.ProducerMessage{
		Topic: TOPIC,
		Value: &message,
	})
	if err != nil {
		log.Println(err)
		return
	}
	value := map[string]string{
		"method": message.Method,
		"url":    message.URL,
		"value":  message.Value,
		"date":   message.Date,
	}
	fmt.Println(fmt.Sprintf("/%d/%d/%+v", partition, offset, value))
}
複製代碼

好比咱們按照上面的配置發送消息：topic: topic-golang partition/offset/value

/0/0/map[date:12344 method:get5 url:www.baidu.com4 value:da4]
/0/1/map[date:12344 method:get5 url:www.baidu.com4 value:da4]
/0/2/map[date:12344 method:get5 url:www.baidu.com4 value:da4]
/0/3/map[date:12344 method:get5 url:www.baidu.com4 value:da4]
複製代碼

上文只有一個 partition , offset值不斷增長。

建立另一個 topic, 分10個區。topic: topic-python

在日誌中顯示成咋樣的呢？

// cd log.dirs  ; server.properties 中的設置

topic-golang-0
topic-python-0
topic-python-1
topic-python-2
topic-python-3
topic-python-4
topic-python-5
topic-python-6
topic-python-7
topic-python-8
topic-python-9
複製代碼

往 topic-python 中發送日誌，分區規則輪詢：

/0/0/map[date:12344 method:get5 url:www.baidu.com4 value:da4]
/1/0/map[date:12344 method:get5 url:www.baidu.com4 value:da4]
/2/0/map[date:12344 method:get5 url:www.baidu.com4 value:da4]
/3/0/map[date:12344 method:get5 url:www.baidu.com4 value:da4]
/4/0/map[date:12344 method:get5 url:www.baidu.com4 value:da4]
/5/0/map[date:12344 method:get5 url:www.baidu.com4 value:da4]
/6/0/map[date:12344 method:get5 url:www.baidu.com4 value:da4]
/7/0/map[date:12344 method:get5 url:www.baidu.com4 value:da4]
/8/0/map[date:12344 method:get5 url:www.baidu.com4 value:da4]
/9/0/map[date:12344 method:get5 url:www.baidu.com4 value:da4]
/0/1/map[date:12344 method:get5 url:www.baidu.com4 value:da4]
/1/1/map[date:12344 method:get5 url:www.baidu.com4 value:da4]
複製代碼

輪詢，不斷的往分區內存消息。

4.2 消費者

系統 B

func main() {
	config := sarama.NewConfig()
	config.Consumer.Return.Errors = true
	brokers := []string{"127.0.0.1:9092"}
	master, err := sarama.NewConsumer(brokers, config)
	if err != nil {
		panic(err)
	}
	defer func() {
		if err := master.Close(); err != nil {
			panic(err)
		}
	}()
	_, e := master.Partitions("topic-python")
	if e != nil {
		log.Println(e)
	}
	consumer, err := master.ConsumePartition("topic-python", 0, sarama.OffsetOldest)
	if err != nil {
		panic(err)
	}
	signals := make(chan os.Signal, 1)
	signal.Notify(signals, os.Interrupt)
	doneCh := make(chan struct{})
	go func() {
		for {
			select {
			case err := <-consumer.Errors():
				fmt.Println(err)
			case msg := <-consumer.Messages():
				fmt.Println("Received messages", string(msg.Key), string(msg.Value), msg.Topic)
			case <-signals:
				fmt.Println("Interrupt is detected")
				doneCh <- struct{}{}
			}
		}
	}()
	<-doneCh
}
複製代碼
  • 消費者指定了 topic: topic-python
  • 消費者指定了 partition: 0

還記得生產者向 topic-python 內發送的消息嗎？ partition/offset/value

/0/0/map[date:12344 method:get5 url:www.baidu.com4 value:da4]
/1/0/map[date:12344 method:get5 url:www.baidu.com4 value:da4]
/2/0/map[date:12344 method:get5 url:www.baidu.com4 value:da4]
/3/0/map[date:12344 method:get5 url:www.baidu.com4 value:da4]
/4/0/map[date:12344 method:get5 url:www.baidu.com4 value:da4]
/5/0/map[date:12344 method:get5 url:www.baidu.com4 value:da4]
/6/0/map[date:12344 method:get5 url:www.baidu.com4 value:da4]
/7/0/map[date:12344 method:get5 url:www.baidu.com4 value:da4]
/8/0/map[date:12344 method:get5 url:www.baidu.com4 value:da4]
/9/0/map[date:12344 method:get5 url:www.baidu.com4 value:da4]
/0/1/map[date:12344 method:get5 url:www.baidu.com4 value:da4]
/1/1/map[date:12344 method:get5 url:www.baidu.com4 value:da4]
複製代碼

能夠看出：partition: 0 中有兩條消息。那麼消費者指定了分區，只能消費這兩條消息。

Received messages  {"method":"get5","url":"www.baidu.com4","value":"da4","date":"12344"} topic-python
Received messages  {"method":"get5","url":"www.baidu.com4","value":"da4","date":"12344"} topic-python
複製代碼

4.3 其餘

使用 kafka 客戶端 ，那麼咱們還須要哪些功能？

  • 關於 Topic 的建立、描述、刪除等
  • 消費者組描述等
  • 元信息：metadata
type ClusterAdmin interface {
	CreateTopic(topic string, detail *TopicDetail, validateOnly bool) error
	ListTopics() (map[string]TopicDetail, error)
	DescribeTopics(topics []string) (metadata []*TopicMetadata, err error)
	DeleteTopic(topic string) error
	CreatePartitions(topic string, count int32, assignment [][]int32, validateOnly bool) error
	DeleteRecords(topic string, partitionOffsets map[int32]int64) error
	DescribeConfig(resource ConfigResource) ([]ConfigEntry, error)
	AlterConfig(resourceType ConfigResourceType, name string, entries map[string]*string, validateOnly bool) error
	CreateACL(resource Resource, acl Acl) error
	ListAcls(filter AclFilter) ([]ResourceAcls, error)
	DeleteACL(filter AclFilter, validateOnly bool) ([]MatchingAcl, error)
	ListConsumerGroups() (map[string]string, error)
	DescribeConsumerGroups(groups []string) ([]*GroupDescription, error)
	ListConsumerGroupOffsets(group string, topicPartitions map[string][]int32) (*OffsetFetchResponse, error)
	DeleteConsumerGroup(group string) error
	DescribeCluster() (brokers []*Broker, controllerID int32, err error)
	Close() error
}
複製代碼

關於單節點 kafka 的基本應用就這些。

5. 容器服務

任何提供服務的系統，均可以使用容器版本，kafka 也可使用容器版本。配置可使用環境變量的形式設置。

docker-compose.yml

version: '2'
services:
 ui:
 image: index.docker.io/sheepkiller/kafka-manager:latest
 depends_on:
 - zookeeper
 ports:
 - 9000:9000
 environment:
 ZK_HOSTS: zookeeper:2181
 zookeeper:
 image: index.docker.io/wurstmeister/zookeeper:latest
 ports:
 - 2181:2181
 server:
 image: index.docker.io/wurstmeister/kafka:latest
 depends_on:
 - zookeeper
 ports:
 - 9092:9092
 environment:
 KAFKA_OFFSETS_TOPIC_REPLIATION_FACTOR: 1
 KAFKA_ADVERTISED_HOST_NAME: 127.0.0.1
 KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
複製代碼
  • zookeeper 分佈式協調系統
  • kafka server Kafka 服務
  • kafka-manager kafka 管理平臺

後續集羣版本。

<完>

代碼：github.com/wuxiaoxiaos…

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程