下載

成都創(chuàng)新互聯(lián)公司是一家專注于成都網(wǎng)站建設(shè)、網(wǎng)站建設(shè)與策劃設(shè)計(jì),南州晴隆網(wǎng)站建設(shè)哪家好?成都創(chuàng)新互聯(lián)公司做網(wǎng)站,專注于網(wǎng)站建設(shè)十余年,網(wǎng)設(shè)計(jì)領(lǐng)域的專業(yè)建站公司;建站業(yè)務(wù)涵蓋:南州晴隆等地區(qū)。南州晴隆做網(wǎng)站價(jià)格咨詢:18982081108

http://kafka.apache.org/downloads.html

http://mirror.bit.edu.cn/apache/kafka/0.11.0.0/kafka_2.11-0.11.0.0.tgz 

root@master:/usr/local/kafka_2.11-0.11.0.0/config# vim server.properties

broker.id=2            每個(gè)節(jié)點(diǎn)不同

log.retention.hours=168

message.max.byte=5242880

default.replication.factor=2

replica.fetch.max.bytes=5242880

zookeeper.connect=master:2181,slave1:2181,slave2:2181

復(fù)制到其它節(jié)點(diǎn)

注意要事先在zk中創(chuàng)建/kafka節(jié)點(diǎn)，否則會(huì)報(bào)出錯(cuò)誤：java.lang.IllegalArgumentException: Path length must be > 0

root@master:/usr/local/zookeeper-3.4.9# bin/zkCli.sh -server master

[zk: master(CONNECTED) 7] create /kafka ''

Created /kafka

[zk: master(CONNECTED) 8] ls /            

[cluster, controller, controller_epoch, brokers, zookeeper, kafka, admin, isr_change_notification, consumers, latest_producer_id_block, config]

[zk: master(CONNECTED) 9] ls /kafka

[]

以守護(hù)進(jìn)程方式啟動(dòng)kafka

root@master:/usr/local/kafka_2.11-0.11.0.0# nohup bin/kafka-server-start.sh config/server.properties &

創(chuàng)建topic：

root@slave2:/usr/local/kafka_2.11-0.11.0.0# bin/kafka-topics.sh  --create  --zookeeper  master:2181  --replication-factor 1  --partitions  1  --topic test

Created topic "test".

列出所有topic：

root@slave2:/usr/local/kafka_2.11-0.11.0.0# bin/kafka-topics.sh --list --zookeeper master:2181

test

發(fā)送消息

root@master:/usr/local/kafka_2.11-0.11.0.0# bin/kafka-console-producer.sh --broker-list master:9092 --topic test

>this is a message

>this is ant^H message

消費(fèi)消息

root@slave1:/usr/local/kafka_2.11-0.11.0.0# bin/kafka-console-consumer.sh --zookeeper master:2181 --topic test --from-beginning

Using the ConsoleConsumer with old consumer is deprecated and will be removed in a future major release. Consider using the new consumer by passing [bootstrap-server] instead of [zookeeper].

this is a message

this is an message

查看集群狀態(tài)信息

root@slave1:/usr/local/kafka_2.11-0.11.0.0# bin/kafka-topics.sh --describe --zookeeper slave1:2181 --topic my-replicated-topic

Topic:my-replicated-topic    PartitionCount:1    ReplicationFactor:3    Configs:

    Topic: my-replicated-topic    Partition: 0    Leader: 3    Replicas: 1,3,2    Isr: 3,2

安裝kafka-manager

root@master:/usr/local/kafka_2.11-0.11.0.0# git clone https://github.com/yahoo/kafka-manager

root@master:/usr/local/kafka_2.11-0.11.0.0/kafka-manager# cd kafka-manager/

root@master:/usr/local/kafka_2.11-0.11.0.0/kafka-manager# ./sbt clean dist

[success] Total time: 3453 s, completed Aug 7, 2017 8:48:15 PM

打包好的文件存在

root@master:/usr/local/kafka_2.11-0.11.0.0/kafka-manager/target/universal# ls

kafka-manager-1.3.3.12.zip  tmp

修改kafka-manager配置文件

root@master:/usr/local/kafka_2.11-0.11.0.0/kafka-manager-1.3.3.12# vim conf/application.conf

kafka-manager.zkhosts="192.168.117.243:2181,192.168.117.45:2181,192.168.117.242:2181"

啟動(dòng)kafka-manager

root@master:/usr/local/kafka_2.11-0.11.0.0/kafka-manager-1.3.3.12# bin/kafka-manager -Dconfig.file=conf/application.conf

推薦啟動(dòng)方式：

root@master:/usr/local/kafka_2.11-0.11.0.0/kafka-manager-1.3.3.12# nohup bin/kafka-manager -Dconfig.file=conf/application.conf -Dhttp.port=7778 &

登錄kafka manager:

http://192.168.117.243:7778/

root@master:/usr/local/kafka_2.11-0.11.0.0/kafka-manager-1.3.3.12# netstat -antlup | grep 7778

tcp6       0      0 :::7778                 :::*                    LISTEN      100620/java   

root@master:/usr/local/kafka_2.11-0.11.0.0/kafka-manager-1.3.3.12# bin/kafka-manager -Dconfig.file=conf/application.conf

This application is already running (Or delete /usr/local/kafka_2.11-0.11.0.0/kafka-manager-1.3.3.12/RUNNING_PID file).

停止kafka-manager

root@master:/usr/local/kafka_2.11-0.11.0.0/kafka-manager-1.3.3.12# rm RUNNING_PID

root@master:/usr/local/kafka_2.11-0.11.0.0# cd kafka-manager-1.0-SNAPSHOT/

生產(chǎn)服務(wù)器配置

# Replication configurations

num.replica.fetchers=4

replica.fetch.max.bytes=1048576

replica.fetch.wait.max.ms=500

replica.high.watermark.checkpoint.interval.ms=5000

replica.socket.timeout.ms=30000

replica.socket.receive.buffer.bytes=65536

replica.lag.time.max.ms=10000

replica.lag.max.messages=4000

controller.socket.timeout.ms=30000

controller.message.queue.size=10

# Log configuration

num.partitions=8

message.max.bytes=1000000

auto.create.topics.enable=true

log.index.interval.bytes=4096

log.index.size.max.bytes=10485760

log.retention.hours=168

log.flush.interval.ms=10000

log.flush.interval.messages=20000

log.flush.scheduler.interval.ms=2000

log.roll.hours=168

log.retention.check.interval.ms=300000

log.segment.bytes=1073741824

# ZK configuration

zookeeper.connection.timeout.ms=6000

zookeeper.sync.time.ms=2000

# Socket server configuration

num.io.threads=8

num.network.threads=8

socket.request.max.bytes=104857600

socket.receive.buffer.bytes=1048576

socket.send.buffer.bytes=1048576

queued.max.requests=16

fetch.purgatory.purge.interval.requests=100

producer.purgatory.purge.interval.requests=100

Kafka是一種高吞吐量的分布式發(fā)布訂閱的消息隊(duì)列系統(tǒng)，原本開(kāi)發(fā)自LinkedIn，用作LinkedIn的活動(dòng)流（ActivityStream）和運(yùn)營(yíng)數(shù)據(jù)處理管道（Pipeline）的基礎(chǔ)。現(xiàn)在它已被多家不同類型的公司作為多種類型的數(shù)據(jù)管道和消息系統(tǒng)使用。

1 Kafka消息隊(duì)列簡(jiǎn)介

1.1 基本術(shù)語(yǔ)

• Broker
  Kafka集群包含一個(gè)或多個(gè)服務(wù)器，這種服務(wù)器被稱為broker[5] 

• Topic
  每條發(fā)布到Kafka集群的消息都有一個(gè)類別，這個(gè)類別被稱為Topic。（物理上不同Topic的消息分開(kāi)存儲(chǔ)，邏輯上一個(gè)Topic的消息雖然保存于一個(gè)或多個(gè)broker上但用戶只需指定消息的Topic即可生產(chǎn)或消費(fèi)數(shù)據(jù)而不必關(guān)心數(shù)據(jù)存于何處）

• Partition
  Partition是物理上的概念，每個(gè)Topic包含一個(gè)或多個(gè)Partition.（一般為kafka節(jié)點(diǎn)數(shù)cpu的總核數(shù)）

• Producer
  負(fù)責(zé)發(fā)布消息到Kafka broker

• Consumer
  消息消費(fèi)者，向Kafka broker讀取消息的客戶端。

• Consumer Group
  每個(gè)Consumer屬于一個(gè)特定的Consumer Group（可為每個(gè)Consumer指定group name，若不指定group name則屬于默認(rèn)的group）。

1.2 消息隊(duì)列

1.2.1 基本特性

1. 可擴(kuò)展

2. 在不需要下線的情況下進(jìn)行擴(kuò)容

3. 數(shù)據(jù)流分區(qū)(partition)存儲(chǔ)在多個(gè)機(jī)器上

4. 高性能

5. 單個(gè)broker就能服務(wù)上千客戶端

6. 單個(gè)broker每秒種讀/寫可達(dá)每秒幾百兆字節(jié)

7. 多個(gè)brokers組成的集群將達(dá)到非常強(qiáng)的吞吐能力

8. 性能穩(wěn)定，無(wú)論數(shù)據(jù)多大

9. Kafka在底層摒棄了Java堆緩存機(jī)制，采用了操作系統(tǒng)級(jí)別的頁(yè)緩存，同時(shí)將隨機(jī)寫操作改為順序?qū)?，再結(jié)合Zero-Copy的特性極大地改善了IO性能。

10. 持久存儲(chǔ)

11. 存儲(chǔ)在磁盤上

12. 冗余備份到其他服務(wù)器上以防止丟失

1.2.2 消息格式

1. 一個(gè)topic對(duì)應(yīng)一種消息格式，因此消息用topic分類

2. 一個(gè)topic代表的消息有1個(gè)或者多個(gè)patition(s)組成

3. 一個(gè)partition中

4. 一個(gè)partition應(yīng)該存放在一到多個(gè)server上

• 一個(gè)server為leader

• 其他servers為followers

• leader需要接受讀寫請(qǐng)求

• followers僅作冗余備份

• leader出現(xiàn)故障，會(huì)自動(dòng)選舉一個(gè)follower作為leader，保證服務(wù)不中斷

• 每個(gè)server都可能扮演一些partitions的leader和其它partitions的follower角色，這樣整個(gè)集群就會(huì)達(dá)到負(fù)載均衡的效果

• 如果只有一個(gè)server，就沒(méi)有冗余備份，是單機(jī)而不是集群

• 如果有多個(gè)server

消息按順序存放消息順序不可變只能追加消息，不能插入每個(gè)消息都有一個(gè)offset，用作消息ID, 在一個(gè)partition中唯一offset有consumer保存和管理，因此讀取順序?qū)嶋H上是完全有consumer決定的，不一定是線性的消息有超時(shí)日期，過(guò)期則刪除

1.2.3 生產(chǎn)者 producer

• producer將消息寫入kafka

• 寫入要指定topic和partition

• 消息如何分到不同的partition，算法由producer指定

1.2.4 消費(fèi)者 consumer

• consumer讀取消息并作處理

• consumer group

• • 可以按照partition的數(shù)目進(jìn)行并發(fā)處理

  • 每個(gè)partition都只有一個(gè)consumer讀取，因而保證了消息被處理的順序是按照partition的存放順序進(jìn)行，注意這個(gè)順序受到producer存放消息的算法影響

  • 這個(gè)概念的引入為了支持兩種場(chǎng)景：每條消息分發(fā)一個(gè)消費(fèi)者，每條消息廣播給所有消費(fèi)者

  • 多個(gè)consumer group訂閱一個(gè)topic，該topci的消息廣播給所有consumer group

  • 一條消息發(fā)送到一個(gè)consumer group后，只能由該group的一個(gè)consumer接收和使用

  • 一個(gè)group中的每個(gè)consumer對(duì)應(yīng)一個(gè)partition可以帶來(lái)如下好處

•  一個(gè)Consumer可以有多個(gè)線程進(jìn)行消費(fèi)，線程數(shù)應(yīng)不多于topic的partition數(shù)，因?yàn)閷?duì)于一個(gè)包含一或多消費(fèi)線程的consumer group來(lái)說(shuō)，一個(gè)partition只能分給其中的一個(gè)消費(fèi)線程消費(fèi)，且讓盡可能多的線程能分配到partition（不過(guò)實(shí)際上真正去消費(fèi)的線程及線程數(shù)還是由線程池的調(diào)度機(jī)制來(lái)決定）。這樣如果線程數(shù)比partition數(shù)多，那么單射分配也會(huì)有多出的線程，它們就不會(huì)消費(fèi)到任何一個(gè)partition的數(shù)據(jù)而空轉(zhuǎn)耗資源 。

• 如果consumer從多個(gè)partition讀到數(shù)據(jù)，不保證數(shù)據(jù)間的順序性，kafka只保證在一個(gè)partition上數(shù)據(jù)是有序的，但多個(gè)partition，根據(jù)你讀的順序會(huì)有不同

• 增減consumer，broker，partition會(huì)導(dǎo)致rebalance，所以rebalance后consumer對(duì)應(yīng)的partition會(huì)發(fā)生變化

新聞標(biāo)題：Ubuntu16.04安裝Kafka集群
文章位置：http://aaarwkj.com/article8/pegdip.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供品牌網(wǎng)站制作、網(wǎng)站內(nèi)鏈、自適應(yīng)網(wǎng)站、企業(yè)網(wǎng)站制作、外貿(mào)網(wǎng)站建設(shè)、服務(wù)器托管

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請(qǐng)盡快告知，我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng)，如需處理請(qǐng)聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時(shí)需注明來(lái)源： 創(chuàng)新互聯(lián)