kafka 基本介紹

創(chuàng)新互聯(lián)公司主要從事成都網(wǎng)站設(shè)計(jì)、網(wǎng)站建設(shè)、網(wǎng)頁(yè)設(shè)計(jì)、企業(yè)做網(wǎng)站、公司建網(wǎng)站等業(yè)務(wù)。立足成都服務(wù)集安,十載網(wǎng)站建設(shè)經(jīng)驗(yàn),價(jià)格優(yōu)惠、服務(wù)專業(yè),歡迎來(lái)電咨詢建站服務(wù):18980820575

 概念

 一個(gè)分布式流處理平臺(tái)，消息訂閱平臺(tái)，一般有三個(gè)特性

1. 可以讓你發(fā)布和訂閱流式的記錄。這一方面與消息隊(duì)列或者企業(yè)消息系統(tǒng)類似。
2. 可以儲(chǔ)存流式的記錄，并且有較好的容錯(cuò)性。
3. 可以在流式記錄產(chǎn)生時(shí)就進(jìn)行處理。

適用場(chǎng)景

1. 構(gòu)造實(shí)時(shí)流數(shù)據(jù)管道，它可以在系統(tǒng)或應(yīng)用之間可靠地獲取數(shù)據(jù)。 (相當(dāng)于message queue)

2. 構(gòu)建實(shí)時(shí)流式應(yīng)用程序，對(duì)這些流數(shù)據(jù)進(jìn)行轉(zhuǎn)換或者影響。 (就是流處理，通過(guò)kafka stream topic和topic之間內(nèi)部進(jìn)行變化)

   部署安裝

kafka 部署安裝需要依賴于 zoookper 和java 環(huán)境

安裝java 環(huán)境

yum install java-1.8.0-openjdk* -y

安裝zoookper

• 下載軟件包

wget https://mirrors.tuna.tsinghua.edu.cn/apache/zookeeper/zookeeper-3.4.14/zookeeper-3.4.14.tar.gz

解壓縮到指定位置

• 修改配置文件

cp zoo_sample.cfg zoo.cfg

配置文件如下，創(chuàng)建所需要的目錄

tickTime=2000
initLimit=10
syncLimit=5
dataDir=/usr/local/zookeeper_data
clientPort=2181

• 制作標(biāo)準(zhǔn)服務(wù)啟動(dòng)

cat /etc/systemd/system/zookeeper.service

[Unit]
Description=zookeeper.service
After=network.target
[Service]
Type=forking
ExecStart=/usr/local/zookeeper/bin/zkServer.sh start
ExecStop=/usr/local/zookeeper/bin/zkServer.sh stop
ExecReload=/usr/local/zookeeper/bin/zkServer.sh restart
[Install]
WantedBy=multi-user.target

systemctl start zookeeper

安裝kafka

curl -LO https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.1.0/kafka_2.12-2.1.0.tgz

解壓縮到指定位置

• 修改配置文件

  server.properties

broker.id=1
port=9092
host.name=localhost
log.dirs=/usr/local/kafka_logs
zookeeper.connect=localhost:2181

zookeeper.properties

dataDir=/usr/local/kafaka_data
clientPort=2181
maxClientCnxns=0
host.name=localhost

 producer.properties

metadata.broker.list=localhost:9092
bootstrap.servers=localhost:9092
compression.type=none

 consumer.properties

bootstrap.servers=localhost:9092
group.id=test-consumer-group
zookeeper.connect=localhost:2181

* 制作標(biāo)準(zhǔn)服務(wù)啟動(dòng)

cat /etc/systemd/system/kafka.service

[Unit]
Description=Apache Kafka server (broker)
After=network.target
After=syslog.target
After=zookeeper.target

[Service]
Type=forking
User=root
Group=root
ExecStart=/usr/local/kafka/bin/kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties
ExecStop=/usr/local/kafka/bin/kafka-server-stop.sh

ExecReload=/bin/kill -HUP $MAINPID
KillMode=none

Restart=on-failure
RestartSec=5s

[Install]
WantedBy=multi-user.target

systemctl start kafka

• 測(cè)試

創(chuàng)建話題Topic
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic kafka01
Created topic "kafka01".

bin/kafka-topics.sh --list --zookeeper localhost:2181

啟動(dòng)消息生產(chǎn)者并發(fā)送消息

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic kafka01

啟動(dòng)消息消費(fèi)者并收到消息

bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic kafka01 --from-beginning

查看topic 列表 詳細(xì)信息

bin/kafka-topics.sh --zookeeper localhost:2181 --list

bin/kafka-topics.sh --zookeeper localhost:2181 --topic kafka01 --describe

Topic:kafka01 PartitionCount:1 ReplicationFactor:1 Configs:
Topic: kafka01 Partition: 0 Leader: 1 Replicas: 1 Isr: 1

拓展分區(qū)

bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic kafka01 --partitions 2

bin/kafka-topics.sh --zookeeper localhost:2181 --topic kafka01 --describe
Topic:kafka01 PartitionCount:2 ReplicationFactor:1 Configs:
Topic: kafka01 Partition: 0 Leader: 1 Replicas: 1 Isr: 1
Topic: kafka01 Partition: 1 Leader: 1 Replicas: 1 Isr: 1