溫馨提示×
您好，登錄后才能下訂單哦！
點(diǎn)擊 登錄注冊 即表示同意《億速云用戶服務(wù)條款》
您好，登錄后才能下訂單哦！
[TOC]
消息 Message
網(wǎng)絡(luò)中的兩臺計(jì)算機(jī)或者兩個(gè)通訊設(shè)備之間傳遞的數(shù)據(jù)。例如說：文本、音樂、視頻等內(nèi)容。
隊(duì)列 Queue
一種特殊的線性表（數(shù)據(jù)元素首尾相接），特殊之處在于只允許在首部刪除元素和在尾部追加元素。入隊(duì)、出隊(duì)。
消息隊(duì)列 MQ
消息+隊(duì)列，保存消息的隊(duì)列。消息的傳輸過程中的容器；主要提供生產(chǎn)、消費(fèi)接口供外部調(diào)用做數(shù)據(jù)的存儲和獲取。MQ主要分為兩類：點(diǎn)對點(diǎn)(p2p)、發(fā)布訂閱(Pub/Sub)
共同點(diǎn)：
消息生產(chǎn)者生產(chǎn)消息發(fā)送到queue中，然后消息消費(fèi)者從queue中讀取并且消費(fèi)消息。
不同點(diǎn)：
p2p模型包括：消息隊(duì)列(Queue)、發(fā)送者(Sender)、接收者(Receiver)
一個(gè)生產(chǎn)者生產(chǎn)的消息只有一個(gè)消費(fèi)者(Consumer)(即一旦被消費(fèi)，消息就不在消息隊(duì)列中)。比如說打電話。
Pub/Sub包含：消息隊(duì)列(Queue)、主題(Topic)、發(fā)布者(Publisher)、訂閱者(Subscriber)
每個(gè)消息可以有多個(gè)消費(fèi)者，彼此互不影響。比如我發(fā)布一個(gè)微博：關(guān)注我的人都能夠看到。
那么在大數(shù)據(jù)領(lǐng)域呢，為了滿足日益增長的數(shù)據(jù)量，也有一款可以滿足百萬級別消息的生成和消費(fèi)，分布式、持久穩(wěn)定的產(chǎn)品——Kafka。 Kafka是分布式的發(fā)布—訂閱消息系統(tǒng)。它最初由LinkedIn(領(lǐng)英)公司發(fā)布，使用Scala語言編寫，與2010年12月份開源，成為Apache的頂級項(xiàng)目。
Kafka是一個(gè)高吞吐量的、持久性的、分布式發(fā)布訂閱消息系統(tǒng)。
它主要用于處理活躍的數(shù)據(jù)(登錄、瀏覽、點(diǎn)擊、分享、喜歡等用戶行為產(chǎn)生的數(shù)據(jù))。
三大特點(diǎn)：
高吞吐量
可以滿足每秒百萬級別消息的生產(chǎn)和消費(fèi)——生產(chǎn)消費(fèi)。QPS
持久性
有一套完善的消息存儲機(jī)制，確保數(shù)據(jù)的高效安全的持久化——中間存儲。
分布式
基于分布式的擴(kuò)展和容錯(cuò)機(jī)制；Kafka的數(shù)據(jù)都會復(fù)制到幾臺服務(wù)器上。當(dāng)某一臺故障失效時(shí)，生產(chǎn)者和消費(fèi)者轉(zhuǎn)而使用其它的機(jī)器——整體健壯性。一個(gè)MQ需要哪些部分？生產(chǎn)、消費(fèi)、消息類別、存儲等等。
對于kafka而言，kafka服務(wù)就像是一個(gè)大的水池。不斷的生產(chǎn)、存儲、消費(fèi)著各種類別的消息。那么kafka由何組成呢？
> Kafka服務(wù)：
> Topic：主題，Kafka處理的消息的不同分類。
> Broker：消息代理，Kafka集群中的一個(gè)kafka服務(wù)節(jié)點(diǎn)稱為一個(gè)broker，主要存儲消息數(shù)據(jù)。存在硬盤中。每個(gè)topic都是有分區(qū)的。
> Partition：Topic物理上的分組，一個(gè)topic在broker中被分為1個(gè)或者多個(gè)partition，分區(qū)在創(chuàng)建topic的時(shí)候指定。
> Message：消息，是通信的基本單位，每個(gè)消息都屬于一個(gè)partition
> Kafka服務(wù)相關(guān)
> Producer：消息和數(shù)據(jù)的生產(chǎn)者，向Kafka的一個(gè)topic發(fā)布消息。
> Consumer：消息和數(shù)據(jù)的消費(fèi)者，定于topic并處理其發(fā)布的消息。
> Zookeeper：協(xié)調(diào)kafka的正常運(yùn)行。Broker：配置文件server.properties
1、為了減少磁盤寫入的次數(shù),broker會將消息暫時(shí)buffer起來,當(dāng)消息的個(gè)數(shù)達(dá)到一定閥值或者過了一定的時(shí)間間隔時(shí),再flush到磁盤,這樣減少了磁盤IO調(diào)用的次數(shù)。
配置：Log Flush Policy
#log.flush.interval.messages=10000 一個(gè)分區(qū)的消息數(shù)閥值
#log.flush.interval.ms=1000
2、kafka的消息保存一定時(shí)間（通常為7天）后會被刪除。
配置：Log Retention Policy
log.retention.hours=168
#log.retention.bytes=1073741824
log.retention.check.interval.ms=300000Producer：配置文件：producer.properties
1、自定義partition
Producer也根據(jù)用戶設(shè)置的算法來根據(jù)消息的key來計(jì)算輸入哪個(gè)partition：partitioner.class
2、異步或者同步發(fā)送
配置項(xiàng)：producer.type
異步或者同步發(fā)送
同步是指：發(fā)送方發(fā)出數(shù)據(jù)后，等接收方發(fā)回響應(yīng)以后才發(fā)下一個(gè)數(shù)據(jù)的通訊方式。
異步是指：發(fā)送方發(fā)出數(shù)據(jù)后，不等接收方發(fā)回響應(yīng)，接著發(fā)送下個(gè)數(shù)據(jù)的通訊方式。
3、批量發(fā)送可以很有效的提高發(fā)送效率。
Kafka producer的異步發(fā)送模式允許進(jìn)行批量發(fā)送，先將消息緩存在內(nèi)存中，然后一次請求批量發(fā)送出去。
具體配置queue.buffering.max.ms、queue.buffering.max.messages。
默認(rèn)值分別為5000和10000consumers：配置文件：consumer.properties
1、每個(gè)consumer屬于一個(gè)consumer group，可以指定組id。group.id
2、消費(fèi)形式：
組內(nèi)：組內(nèi)的消費(fèi)者消費(fèi)同一份數(shù)據(jù)；同時(shí)只能有一個(gè)consumer消費(fèi)一個(gè)Topic中的1個(gè)partition；一個(gè)consumer可以消費(fèi)多個(gè)partitions中的消息。
所以，對于一個(gè)topic,同一個(gè)group中推薦不能有多于partitions個(gè)數(shù)的consumer同時(shí)消費(fèi),否則將意味著某些consumer將無法得到消息。
組間：每個(gè)消費(fèi)組消費(fèi)相同的數(shù)據(jù)，互不影響。
3、在一個(gè)consumer多個(gè)線程的情況下，一個(gè)線程相當(dāng)于一個(gè)消費(fèi)者。
例如：partition為3，一個(gè)consumer起了3個(gè)線程消費(fèi)，另一個(gè)后來的consumer就無法消費(fèi)。
（這是Kafka用來實(shí)現(xiàn)一個(gè)Topic消息的廣播（發(fā)給所有的Consumer）和單播（發(fā)給某一個(gè)Consumer）的手段。
一個(gè)Topic可以對應(yīng)多個(gè)Consumer Group。如果需要實(shí)現(xiàn)廣播，只要每個(gè)Consumer有一個(gè)獨(dú)立的Group就可以了。
要實(shí)現(xiàn)單播只要所有的Consumer在同一個(gè)Group里。用Consumer Group還可以將Consumer進(jìn)行自由的分組而不需要多次發(fā)送消息到不同的Topic。）1、每個(gè)partition在存儲層面是append log文件。新消息都會被直接追加到log文件的尾部，每條消息在log文件中的位置稱為offset（偏移量）。
2、每條Message包含了以下三個(gè)屬性：
1°、offset 對應(yīng)類型：long 此消息在一個(gè)partition中序號?？梢哉J(rèn)為offset是partition中Message的id
2°、MessageSize 對應(yīng)類型：int32 此消息的字節(jié)大小。
3°、data 是message的具體內(nèi)容。
3、越多的partitions意味著可以容納更多的consumer,有效提升并發(fā)消費(fèi)的能力。
4、總之：業(yè)務(wù)區(qū)分增加topic、數(shù)據(jù)量大增加partition。 解壓： [uplooking@uplooking01 ~]$ tar -zxvf soft/kafka_2.10-0.10.0.1.tgz -C app/
重命名：[uplooking@uplooking01 ~]$ mv app/kafka_2.10-0.10.0.1/ app/kafka
添加KAFKA_HOME至環(huán)境變量：~/.bash_profile
export KAFKA_HOME=/home/uplooking/app/kafka
export PATH=$PATH:$KAFKA_HOME/bin
source ~/.bash_profile
配置相關(guān)參數(shù)：$KAFKA_HOME/config/server.properties
主要參數(shù)：broker.id、log.dirs、zookeeper.connect
broker.id=10
log.dirs=/home/uplooking/data/kafka [kafka數(shù)據(jù)的存放目錄]
zookeeper.connect=uplooking01:2181,uplooking02:2181,uplooking03:2181
kafka實(shí)例broker監(jiān)聽默認(rèn)端口9092，配置listeners=PLAINTEXT://:9092
啟動(dòng)：
$KAFKA_HOME/bin/kafka-server-start.sh [-daemon] $KAFKA_HOME/config/server.properties
-daemon 可選，表示后臺啟動(dòng)kafka服務(wù)當(dāng)然，kafka的配置文件也非常重要，有必要對其中的內(nèi)容學(xué)習(xí)一下，這里給出其配置文件的說明：
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# see kafka.server.KafkaConfig for additional details and defaults
############################# Server Basics #############################
##################################################################################
# broker就是一個(gè)kafka的部署實(shí)例，在一個(gè)kafka集群中，每一臺kafka都要有一個(gè)broker.id
# 并且，該id唯一，且必須為整數(shù)
##################################################################################
broker.id=10
############################# Socket Server Settings #############################
# The address the socket server listens on. It will get the value returned from
# java.net.InetAddress.getCanonicalHostName() if not configured.
# FORMAT:
# listeners = security_protocol://host_name:port
# EXAMPLE:
# listeners = PLAINTEXT://your.host.name:9092
#listeners=PLAINTEXT://:9092
# Hostname and port the broker will advertise to producers and consumers. If not set,
# it uses the value for "listeners" if configured. Otherwise, it will use the value
# returned from java.net.InetAddress.getCanonicalHostName().
#advertised.listeners=PLAINTEXT://your.host.name:9092
##################################################################################
#The number of threads handling network requests
# 默認(rèn)處理網(wǎng)絡(luò)請求的線程個(gè)數(shù) 3個(gè)
##################################################################################
num.network.threads=3
##################################################################################
# The number of threads doing disk I/O
# 執(zhí)行磁盤IO操作的默認(rèn)線程個(gè)數(shù) 8
##################################################################################
num.io.threads=8
##################################################################################
# The send buffer (SO_SNDBUF) used by the socket server
# socket服務(wù)使用的進(jìn)行發(fā)送數(shù)據(jù)的緩沖區(qū)大小，默認(rèn)100kb
##################################################################################
socket.send.buffer.bytes=102400
##################################################################################
# The receive buffer (SO_SNDBUF) used by the socket server
# socket服務(wù)使用的進(jìn)行接受數(shù)據(jù)的緩沖區(qū)大小，默認(rèn)100kb
##################################################################################
socket.receive.buffer.bytes=102400
##################################################################################
# The maximum size of a request that the socket server will accept (protection against OOM)
# socket服務(wù)所能夠接受的最大的請求量，防止出現(xiàn)OOM(Out of memory)內(nèi)存溢出，默認(rèn)值為：100m
# （應(yīng)該是socker server所能接受的一個(gè)請求的最大大小，默認(rèn)為100M）
##################################################################################
socket.request.max.bytes=104857600
############################# Log Basics （數(shù)據(jù)相關(guān)部分，kafka的數(shù)據(jù)稱為log）#############################
##################################################################################
# A comma seperated list of directories under which to store log files
# 一個(gè)用逗號分隔的目錄列表，用于存儲kafka接受到的數(shù)據(jù)
##################################################################################
log.dirs=/home/uplooking/data/kafka
##################################################################################
# The default number of log partitions per topic. More partitions allow greater
# parallelism for consumption, but this will also result in more files across
# the brokers.
# 每一個(gè)topic所對應(yīng)的log的partition分區(qū)數(shù)目，默認(rèn)1個(gè)。更多的partition數(shù)目會提高消費(fèi)
# 并行度，但是也會導(dǎo)致在kafka集群中有更多的文件進(jìn)行傳輸
# （partition就是分布式存儲，相當(dāng)于是把一份數(shù)據(jù)分開幾份來進(jìn)行存儲，即劃分塊、劃分分區(qū)的意思）
##################################################################################
num.partitions=1
##################################################################################
# The number of threads per data directory to be used for log recovery at startup and flushing at shutdown.
# This value is recommended to be increased for installations with data dirs located in RAID array.
# 每一個(gè)數(shù)據(jù)目錄用于在啟動(dòng)kafka時(shí)恢復(fù)數(shù)據(jù)和在關(guān)閉時(shí)刷新數(shù)據(jù)的線程個(gè)數(shù)。如果kafka數(shù)據(jù)存儲在磁盤陣列中
# 建議此值可以調(diào)整更大。
##################################################################################
num.recovery.threads.per.data.dir=1
############################# Log Flush Policy （數(shù)據(jù)刷新策略）#############################
# Messages are immediately written to the filesystem but by default we only fsync() to sync
# the OS cache lazily. The following configurations control the flush of data to disk.
# There are a few important trade-offs（平衡） here:
# 1. Durability 持久性: Unflushed data may be lost if you are not using replication.
# 2. Latency 延時(shí)性: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush.
# 3. Throughput 吞吐量: The flush is generally the most expensive operation, and a small flush interval may lead to exceessive seeks.
# The settings below allow one to configure the flush policy to flush data after a period of time or
# every N messages (or both). This can be done globally and overridden on a per-topic basis.
# kafka中只有基于消息條數(shù)和時(shí)間間隔數(shù)來制定數(shù)據(jù)刷新策略，而沒有大小的選項(xiàng)，這兩個(gè)選項(xiàng)可以選擇配置一個(gè)
# 當(dāng)然也可以兩個(gè)都配置，默認(rèn)情況下兩個(gè)都配置，配置如下。
# The number of messages to accept before forcing a flush of data to disk
# 消息刷新到磁盤中的消息條數(shù)閾值
#log.flush.interval.messages=10000
# The maximum amount of time a message can sit in a log before we force a flush
# 消息刷新到磁盤生成一個(gè)log數(shù)據(jù)文件的時(shí)間間隔
#log.flush.interval.ms=1000
############################# Log Retention Policy（數(shù)據(jù)保留策略） #############################
# The following configurations control the disposal（清理） of log segments（分片）. The policy can
# be set to delete segments after a period of time, or after a given size has accumulated（累積）.
# A segment will be deleted whenever（無論什么時(shí)間） *either* of these criteria（標(biāo)準(zhǔn)） are met. Deletion always happens
# from the end of the log.
# 下面的配置用于控制數(shù)據(jù)片段的清理，只要滿足其中一個(gè)策略（基于時(shí)間或基于大?。制蜁粍h除
# The minimum age of a log file to be eligible for deletion
# 基于時(shí)間的策略，刪除日志數(shù)據(jù)的時(shí)間，默認(rèn)保存7天
log.retention.hours=168
# A size-based retention policy for logs. Segments are pruned from the log as long as the remaining
# segments don't drop below log.retention.bytes. 1G
# 基于大小的策略，1G
#log.retention.bytes=1073741824
# The maximum size of a log segment file. When this size is reached a new log segment will be created.
# 數(shù)據(jù)分片策略
log.segment.bytes=1073741824
# The interval at which log segments are checked to see if they can be deleted according
# to the retention policies 5分鐘
# 每隔多長時(shí)間檢測數(shù)據(jù)是否達(dá)到刪除條件
log.retention.check.interval.ms=300000
############################# Zookeeper #############################
# Zookeeper connection string (see zookeeper docs for details).
# This is a comma separated host:port pairs, each corresponding to a zk
# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
# You can also append an optional chroot string to the urls to specify the
# root directory for all kafka znodes.
zookeeper.connect=uplooking01:2181,uplooking02:2181,uplooking03:2181
# Timeout in ms for connecting to zookeeper
zookeeper.connection.timeout.ms=6000另外需要注意的是，kafka啟動(dòng)后，會在zookeeper中創(chuàng)建相關(guān)的節(jié)點(diǎn)：
[zk: localhost:2181(CONNECTED) 1] ls /
[controller, brokers, zookeeper, yarn-leader-election, hadoop-ha, admin, isr_change_notification, consumers, config, hbase]
[zk: localhost:2181(CONNECTED) 5] get /brokers/ids/10
{"jmx_port":-1,"timestamp":"1521936591128","endpoints":["PLAINTEXT://uplooking01:9092"],"host":"uplooking01","version":3,"port":9092}
cZxid = 0xa000000c1
ctime = Sun Mar 25 08:09:50 CST 2018
mZxid = 0xa000000c1
mtime = Sun Mar 25 08:09:50 CST 2018
pZxid = 0xa000000c1
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x6762543b71390005
dataLength = 133
numChildren = 0創(chuàng)建Topic：
kafka-topics.sh --create --topic hadoop --zookeeper uplooking01:2181,uplooking02:2181,uplooking03:2181 --partitions 1 --replication-factor 1
kafka-topics.sh --create --topic hive --zookeeper uplooking01:2181 --partitions 1 --replication-factor 1
kafka-topics.sh --create --topic hbase --zookeeper uplooking01:2181 --partitions 3 --replication-factor 1
創(chuàng)建topic過程的問題，replication-factor個(gè)數(shù)不能超過broker的個(gè)數(shù)
bin/kafka-topics.sh --create --topic sqoop --zookeeper uplooking01:2181 --partitions 3 --replication-factor 3
Error while executing topic command : replication factor: 3 larger than available brokers: 1
另外，在創(chuàng)建topic后，可以在/home/uplooking/data/kafka目錄查看到分區(qū)的目錄，
有多少個(gè)分區(qū)就會相應(yīng)創(chuàng)建多少個(gè)目錄。uplooking01:2181,uplooking02:2181,uplooking03:2181 可以只寫一個(gè)，為了做筆記方便，后面只寫一個(gè)。
查看Topic列表：
kafka-topics.sh --list --zookeeper uplooking01:2181查看某一個(gè)具體的Topic：
kafka-topics.sh --describe xxx --zookeeper uplooking01:2181
Topic:xxx PartitionCount:3 ReplicationFactor:1 Configs:
Topic: xxx Partition: 0 Leader: 10 Replicas: 10 Isr: 10
Topic: xxx Partition: 1 Leader: 10 Replicas: 10 Isr: 10
Topic: xxx Partition: 2 Leader: 10 Replicas: 10 Isr: 10
PartitionCount：topic對應(yīng)的partition的個(gè)數(shù)
ReplicationFactor：topic對應(yīng)的副本因子，說白就是副本個(gè)數(shù)
Partition：partition編號，從0開始遞增
Leader：當(dāng)前partition起作用的breaker.id
Replicas: 當(dāng)前副本數(shù)據(jù)坐在的breaker.id，是一個(gè)列表，排在最前面的起作用
Isr：當(dāng)前kakfa集群中可用的breaker.id列表 修改Topic：
不能修改replication-factor，以及只能對partition個(gè)數(shù)進(jìn)行增加，不能減少
kafka-topics.sh --alter --topic hive --zookeeper uplooking01:2181 --partitions 3
partition由3變?yōu)?的時(shí)，拋出的異常：
ERROR kafka.admin.AdminOperationException: The number of partitions for a topic can only be increased刪除Topic：
kafka-topics.sh --delete --topic hbase --zookeeper uplooking01:2181
Topic hbase is marked for deletion.
Note: This will have no impact if delete.topic.enable is not set to true.
徹底刪除一個(gè)topic，需要在server.properties中配置delete.topic.enable=true，否則只是標(biāo)記刪除
配置完成之后，需要重啟kafka服務(wù)。使用kafka提供的標(biāo)準(zhǔn)生產(chǎn)消費(fèi)腳本。
生產(chǎn)數(shù)據(jù)：
kafka-console-producer.sh --broker-list uplooking01:9092 --topic hadoop
生產(chǎn)數(shù)據(jù)的時(shí)候需要指定：當(dāng)前數(shù)據(jù)流向哪個(gè)broker，以及哪一個(gè)topic消費(fèi)數(shù)據(jù)：
kafka-console-consumer.sh --topic hadoop --zookeeper uplooking01:2181
說明：該消費(fèi)語句，只能獲取最新的數(shù)據(jù)，要想歷史數(shù)據(jù)，需要添加選項(xiàng)--from-beginning
/kafka-console-consumer.sh --topic hadoop --zookeeper uplooking01:2181 --from-beginning
在消費(fèi)數(shù)據(jù)的時(shí)候，只需要指定topic，以及topic的元數(shù)據(jù)信息即可(在ZK中存放)，所以這里需要使用zk消費(fèi)者--黑名單(blacklist)和白名單(whitelist)選項(xiàng)：
--blacklist 后面跟需要過濾的topic的列表，使用","隔開，意思是除了列表中的topic之外，都能接收其它topic的數(shù)據(jù)
--whitelist 后面跟需要過濾的topic的列表，使用","隔開，意思是除了列表中的topic之外，都不能接收其它topic的數(shù)據(jù)
eg:
kafka-console-consumer.sh --zookeeper uplooking01:2181 --from-beginning --blacklist hadoop,hive
kafka-console-consumer.sh --zookeeper uplooking01:2181 --from-beginning --whitelist hadoop,flumekafka中沒有主從節(jié)點(diǎn)的概念，因此只需要將kafka安裝目錄拷貝到其它節(jié)點(diǎn)上即可，不過需要注意的是，需要修改brokerId為唯一的：
scp -r /home/uplooking/app/kafka/ uplooking@uplooking02:/home/uplooking/app
scp -r /home/uplooking/app/kafka/ uplooking@uplooking03:/home/uplooking/app為了方便后面理解kafka的相關(guān)概念，這里將uplooking01、uplooking02、uplooking03的brokerId分別修改為101、102、103.
在三個(gè)節(jié)點(diǎn)上分別啟動(dòng)kafka：
kafka-server-start.sh -daemon app/kafka/config/server.properties創(chuàng)建一個(gè)topic：
kafka-topics.sh --create --topic hadoop --partitions 3 --replication-factor 3 --zookeeper uplooking01:2181,uplooking02:2181,uplooking03:2181查看該topic的詳細(xì)信息：
[uplooking@uplooking01 ~]$ kafka-topics.sh --describe hbase --zookeeper uplooking01:2181,uplooking02:2181,uplooking03:2181
Topic:hadoop PartitionCount:3 ReplicationFactor:3 Configs:
Topic: hadoop Partition: 0 Leader: 101 Replicas: 101,102,103 Isr: 101,102,103
Topic: hadoop Partition: 1 Leader: 102 Replicas: 102,103,101 Isr: 102,103,101
Topic: hadoop Partition: 2 Leader: 103 Replicas: 103,101,102 Isr: 103,101,102
再查看前面的解釋：
PartitionCount：topic對應(yīng)的partition的個(gè)數(shù)
ReplicationFactor：topic對應(yīng)的副本因子，說白就是副本個(gè)數(shù)
Partition：partition編號，從0開始遞增
Leader：當(dāng)前partition起作用的breaker.id
Replicas: 當(dāng)前副本數(shù)據(jù)坐在的breaker.id，是一個(gè)列表，排在最前面的起作用
Isr：當(dāng)前kakfa集群中可用的breaker.id列表
這樣就很容易理解了。這意味著，三個(gè)分區(qū)在三個(gè)節(jié)點(diǎn)上都有保存數(shù)據(jù)的，可以分別在每個(gè)節(jié)點(diǎn)上查看相關(guān)的分區(qū)數(shù)據(jù)信息：
[uplooking@uplooking01 kafka]$ ll
總用量 24
-rw-rw-r-- 1 uplooking uplooking 0 3月 25 19:33 cleaner-offset-checkpoint
drwxrwxr-x 2 uplooking uplooking 4096 3月 25 19:33 hadoop-0
drwxrwxr-x 2 uplooking uplooking 4096 3月 25 19:33 hadoop-1
drwxrwxr-x 2 uplooking uplooking 4096 3月 25 19:33 hadoop-2
-rw-rw-r-- 1 uplooking uplooking 56 3月 25 19:33 meta.properties
-rw-rw-r-- 1 uplooking uplooking 37 3月 25 19:48 recovery-point-offset-checkpoint
-rw-rw-r-- 1 uplooking uplooking 37 3月 25 19:49 replication-offset-checkpoint為了進(jìn)一步理解相關(guān)概念，可以嘗試把uplooking01上的kafka關(guān)掉，然后再查看topic的詳細(xì)信息：
[uplooking@uplooking01 ~]$ kafka-topics.sh --describe hadoop --zookeeper uplooking01:2181,uplooking02:2181,uplooking03:2181
Topic:hadoop PartitionCount:3 ReplicationFactor:3 Configs:
Topic: hadoop Partition: 0 Leader: 102 Replicas: 101,102,103 Isr: 102,103
Topic: hadoop Partition: 1 Leader: 102 Replicas: 102,103,101 Isr: 102,103
Topic: hadoop Partition: 2 Leader: 103 Replicas: 103,101,102 Isr: 103,102然后個(gè)人分析如下：
前面提到：業(yè)務(wù)區(qū)分增加topic、數(shù)據(jù)量大增加partition。
所以partition分區(qū)，是為了把數(shù)據(jù)分散來存放，這好比日志需要每天分割一樣，也是避免單個(gè)存儲位置數(shù)據(jù)量過多。
顯然，至于我們的每個(gè)消息存放在哪個(gè)分區(qū)，kafka本身是有機(jī)制去進(jìn)行計(jì)算的：
int hashCode = Math.abs("ttt".hashCode());
int partition = hashCode % 50;
具體這里就不進(jìn)行討論了。
另外，因?yàn)樵O(shè)置了3個(gè)副本因子，所以3個(gè)分區(qū)的數(shù)據(jù)在3個(gè)節(jié)點(diǎn)上都會有保存，同時(shí)為了起到負(fù)載均衡的作用，kafka
會為每個(gè)分區(qū)設(shè)置一個(gè)leader節(jié)點(diǎn)來專門進(jìn)行該分區(qū)數(shù)據(jù)的相關(guān)操作。
現(xiàn)在再去看前面kafka組件的理論知識，就很容易理解了。
如上圖所示，一般的，Kafka生產(chǎn)的數(shù)據(jù)，是由Flume的Sink提供的，這里我們需要用到Flume集群，
通過Flume集群將Agent的日志收集分發(fā)到 Kafka（供實(shí)時(shí)計(jì)算處理）和HDFS（離線計(jì)算處理）。
這里，我們使用Flume作為日志收集系統(tǒng)，將收集到的數(shù)據(jù)輸送到Kafka中間件，以供Storm去實(shí)時(shí)消費(fèi)計(jì)算，整個(gè)流程從各個(gè)Web節(jié)點(diǎn) 上，
通過Flume的Agent代理收集日志，然后匯總到Flume集群，在由Flume的Sink將日志輸送到Kafka集群，完成數(shù)據(jù)的生產(chǎn)流程。先創(chuàng)建一個(gè)topic：
[uplooking@uplooking01 ~]$ kafka-topics.sh --create --topic flume-kafka --partitions 3 --replication-factor 3 --zookeeper uplooking01:2181,uplooking02:2181,uplooking03:2181
Created topic "flume-kafka".
[uplooking@uplooking01 ~]$ kafka-topics.sh --describe flume-kafka --zookeeper uplooking01:2181,uplooking02:2181,uplooking03:2181
Topic:flume-kafka PartitionCount:3 ReplicationFactor:3 Configs:
Topic: flume-kafka Partition: 0 Leader: 101 Replicas: 101,102,103 Isr: 101,102,103
Topic: flume-kafka Partition: 1 Leader: 102 Replicas: 102,103,101 Isr: 102,103,101
Topic: flume-kafka Partition: 2 Leader: 103 Replicas: 103,101,102 Isr: 103,101,102啟動(dòng)kafka消費(fèi)者：
[uplooking@uplooking01 ~]$ kafka-console-consumer.sh --topic flume-kafka --zookeeper uplooking01:2181,uplooking02:2181,uplooking03:2181Flume的配置文件，這里為監(jiān)聽一個(gè)目錄下的文件變化：
#########################################################
##
##主要作用是監(jiān)聽目錄中的新增文件，采集到數(shù)據(jù)之后，輸出到kafka
## 注意：Flume agent的運(yùn)行，主要就是配置source channel sink
## 下面的a1就是agent的代號，source叫r1 channel叫c1 sink叫k1
#########################################################
a1.sources = r1
a1.sinks = k1
a1.channels = c1
#對于source的配置描述 監(jiān)聽目錄中的新增文件
a1.sources.r1.type = spooldir
a1.sources.r1.spoolDir = /home/uplooking/data/flume/source
a1.sources.r1.fileHeader = true
a1.sources.r1.fileHeaderKey = filepath
a1.sources.r1.fileSuffix = .OK
a1.sources.r1.deletePolicy = immediate
#對于sink的配置描述 使用kafka做數(shù)據(jù)的消費(fèi)
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.topic = flume-kafka
a1.sinks.k1.brokerList = uplooking01:9092,uplooking02:9092,uplooking03:9092
a1.sinks.k1.requiredAcks = 1
a1.sinks.k1.batchSize = 20
#對于channel的配置描述 使用內(nèi)存緩沖區(qū)域做數(shù)據(jù)的臨時(shí)緩存
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
#通過channel c1將source r1和sink k1關(guān)聯(lián)起來
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
啟動(dòng)Flume：
flume-ng agent --conf conf --name a1 --conf-file conf/flume-kafka.conf 向被偵聽目錄中添加hello文件，其內(nèi)容如下：
hello he
hello me
hello you添加后查看kafka消費(fèi)者端的輸出：
[uplooking@uplooking01 ~]$ kafka-console-consumer.sh --topic flume-kafka --zookeeper uplooking01:2181,uplooking02:2181,uplooking03:2181
hello he
hello me
hello you這樣就完成了Kafka和Flume的整合。
免責(zé)聲明：本站發(fā)布的內(nèi)容（圖片、視頻和文字）以原創(chuàng)、轉(zhuǎn)載和分享為主，文章觀點(diǎn)不代表本網(wǎng)站立場，如果涉及侵權(quán)請聯(lián)系站長郵箱：is@yisu.com進(jìn)行舉報(bào)，并提供相關(guān)證據(jù)，一經(jīng)查實(shí)，將立刻刪除涉嫌侵權(quán)內(nèi)容。
億速云公眾號
手機(jī)網(wǎng)站二維碼