Kafka

1 概述

分布式发布订阅模式的消息队列
主要用于大数据领域
解耦,削峰
spark 90%以上数据来源为kafka
更多处理生产大于消费的问题

两种模式

点对点模式
发布订阅模式
问题: 下游消费者处理能力不定
发布订阅模式2种,队列推,消费者拉
kafka为消费者拉,缺点为

架构

生产者 Kafka集群 消费者 ZK
topic 将数据分类
partition 提高topic负载能力
leader 针对topic,follower相当于备份
topic可以只有一个partition
consumer group 某个分区只能被同一消费者组里的某个消费者消费,提高消费能力
zk管理kafka集群
消费者挂掉后重启可以继续消费,需要存消费位置,保存在zk中
0.9版本之前消费偏移量存在zk中,0.9之后存在kafka集群中,避免与zk交互太频繁
kafka存消息存在磁盘,默认存7天

2 安装

准备3台机器node-01 node-02 node-03
分别安装zk并启动,参考:
https://www.jianshu.com/p/eb4d1819c5cf

tar -zxvf kafka_2.11-0.11.0.0.tgz -C /opt/
mv kafka_2.11-0.11.0.0/ kafka
# 修改配置
vi /opt/kafka/config/server.properties
--------------
# 三台机器分别为0 1 2
broker.id=2
# 允许删除主题
delete.topic.enable=true
# 数据目录
log.dirs=/opt/kafka/logs
# zk地址
zookeeper.connect=node-01:2181,node-02:2181,node-03:2181
---------------
export ZOOKEEPER_HOME=/opt/zookeeper
PATH=$PATH:$JAVA_HOME/bin:$HADOOP_HOME/bin:$HADOOP_HOME/sbin:$ZOOKEEPER_HOME/bin
# 分发kafka到另外两台机器,并修改broker.id
xsync
# 3台分别启动kafka
bin/kafka-server-start.sh -daemon config/server.properties
# 查看
jps

# 群启脚本
kk.sh
------------
# !/bin/bash

case $1 in 
"start"){
  for i in node-01 node-02 node-03
  do
    echo "***********$i*************"
    ssh $i "/opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties"
  done
});;

"stop"){
  for i in node-01 node-02 node-03
  do
    echo "***********$i*************"
    ssh $i "/opt/kafka/bin/kafka-server-stop.sh
  done
});;

esac
bin/kafka-server-start.sh -daemon config/server.properties
------------
# 群启
kk.sh start
# 群关
kk.sh stop
# 创建主题 指定zk 分区 副本数
bin/kafka-topics.sh --create --zookeeper node-01:2181 --topic first --partitions 2 --replication-factor 2
# 查看主题
bin/kafka-topics.sh --list --zookeeper node-01:2181
# 删除
bin/kafka-topics.sh --delete --zookeeper node-01:2181 --topic first
# 查看主题信息
bin/kafka-topics.sh --describe --topic first --zookeeper node-01:2181
# 副本数不能超过可用机器数
# node-01 启动生产者
bin/kafka-console-producer.sh --topic first --broker-list node-01:9092
# node-02 启动消费者
# 该方式废弃
bin/kafka-console-consumer.sh --topic first --zookeeper node-02:2181
bin/kafka-console-consumer.sh --topic first --bootstrap-server node-01:9092
# 生产者发消息消费者即可收到
# node-03 启动消费者 指定消费位置
bin/kafka-console-consumer.sh --topic first --zookeeper node-02:2181 --from-beginning