Apache Kafka 企业级消息队列
爬虫课程:原生队列、多线程重复消费的问题、ArrayBlockingQueue阻塞队列
分布式爬虫:使用Redis的list数据结构做队列。
分布式电商:AMQ 消息队列、发布一个商品时发送一个消息,有程序消费消息创建静态化页面。
ApacheKafka:消息队列、随着大数据兴起,现在非常流行。
1、课程目标
- 理解 Apache Kafka是什么
- 掌握Apache Kafka的基本架构
- 搭建Kafka集群
- 掌握操作集群的两种方式
- 熟悉Apache Kafka原理
- Kafka原理-分片与副本机制
- Kafka原理-消息不丢失机制
- Kafka原理-消息存储及查询机制
- Kafka原理-生产者数据分发策略
- Kafka原理-消费者的负载均衡机制
- 了解Apache Kafka 监控及运维
2、Apache Kafka是什么
是什么?有什么用?怎么用?
是什么?三个定义
- Apache Kafka 是一个消息队列(生产者消费者模式)
- Apache Kafka 目标:构建企业中统一的、高通量、低延时的消息平台。
- 大多的是消息队列(消息中间件)都是基于JMS标准实现的,Apache Kafka 类似于JMS的实现。
有什么用?(消息队列有什么用?)
- 作为缓冲,来异构、解耦系统。
- 用户注册需要完成多个步骤,每个步骤执行都需要很长时间。代表用户等待时间是所有步骤的累计时间。
- 为了减少用户等待的时间,使用并行执行,有多少个步骤,就开启多少个线程来执行。代表用户等待时间是所有步骤中耗时最长的那个步骤时间。
- 有了新得问题:开启多线程执行每个步骤,如果以一个步骤执行异常,或者严重超时,用户等待的时间就不可控了。
- 通过消息队列来保证。
- 注册时,立即返回成功。
- 发送注册成功的消息到消息平台。
- 对注册信息感兴趣的程序,可以消息消息。
3、Apache Kafka的基本架构
KafkaCluster:由多个服务器组成。每个服务器单独的名字broker(掮客)。
KafkaProducer:生产者、负责生产数据。
Kafkaconsumer:消费者、负责消费数据。
KafkaTopic: 主题,一类消息的名称。存储数据时将一类数据存放在某个topci下,消费数据也是消费一类数据。
订单系统:创建一个topic,叫做order。
用户系统:创建一个topic,叫做user。
商品系统:创建一个topic,叫做product。
注意:Kafka的元数据都是存放在zookeeper中。
4、搭建Kafka集群
4.1、准备3台虚拟机
192.168.140.128 kafka01
192.168.140.129 kafka02
192.168.140.130 kafka03
4.2、初始化环境
1)安装jdk、安装zookeeper
2)安装目录
安装包存放的目录:/export/software
安装程序存放的目录:/export/servers
数据目录:/export/data
日志目录:/export/logs
mkdir -p /export/servers/
mkdir -p /export/software /
mkdir -p /export/data/
mkdir -p /export/logs /
|
3)安装用户
安装hadoop,会创建一个hadoop用户
安装kafka,创建一个kafka用户
或者创建bigdata用户,用来安装所有的大数据软件。
本例:使用root用户
- 验证环境
- jdk环境
-
- zookeeper环境
zkServer.sh status
4.3、搭建Kafka集群
4.3.1、准备安装包
由于kafka是scala语言编写的,基于scala的多个版本,kafka发布了多个版本。
其中2.11是推荐版本。
4.3.2、下载安装包及解压
tar -zxvf kafka_2.11-1.0.0.tgz -C /export/servers/
cd /export/servers/
rm -rf /export/servers/kafka
rm -rf /export/logs/kafka/
rm -rf /export/data/kafka
mv kafka_2.11-1.0.0 kafka
|
- 解压文件
- 删除之前的安装记录
- 重命名
4.3.3、查看目录及修改配置文件
4.3.3.1查看目录
4.3.3.2修改配置文件
进入配置目录,查看server.properties文件
cat server.properties |grep -v "#"
通过以上命令,查看到了默认的配置文件,对默认的文件进行修改。
修改三个地方
- Borker.id
- 数据存放的目录,注意目录如果不存在,需要新建下。
- zookeeper的地址信息
# broker.id 标识了kafka集群中一个唯一broker。
broker.id=0
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
# 存放生产者生产的数据 数据一般以topic的方式存放
# 创建一个数据存放目录 /export/data/kafka --- mkdir -p /export/data/kafka
log.dirs=/export/data/kafka
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
# zk的信息
zookeeper.connect=zk01:2181,zk02:2181,zk03:2181
zookeeper.connection.timeout.ms=6000
group.initial.rebalance.delay.ms=0
|
4.3.4、分发配置文件及修改brokerid
将修改好的配置文件,分发到node02,node03上。
先在node02、node03上删除以往的安装记录
rm -rf /export/servers/kafka
rm -rf /export/logs/kafka/
rm -rf /export/data/kafka
|
分发安装包
scp -r /export/servers/kafka/ node02:/export/servers/
scp -r /export/servers/kafka/ node03:/export/servers/
|
修改node02上的broker.id
vi /export/servers/kafka/config/server.properties
|
修改node03上的broker.id
vi /export/servers/kafka/config/server.properties
|
4.3.4、启动集群
cd /export/servers/kafka/bin
./kafka-server-start.sh /export/servers/kafka/config/server.properties
|
4.3.5、查看Kafka集群
由于kafka集群并没有UI界面可以查看。
需要借助外部工具,来查看卡夫卡的集群
这个工具是一个java程序,必须要安装好JDK
5、操作集群的两种方式
需求:订单系统,需要发送消息。 后面后3个程序需要接受这个消息,并做后续的处理。
5.1、使用控制台运行
- 创建一个订单的topic。
bin/kafka-topics.sh --create --zookeeper zk01:2181 --replication-factor 1 --partitions 1 --topic order
|
- 编写代码启动一个生产者,生产数据
bin/kafka-console-producer.sh --broker-list kafka01:9092 --topic order
|
- 编写代码启动给一个消费者,消费数据
bin/kafka-console-consumer.sh --zookeeper zk01:2181 --from-beginning --topic order
|
5.2、使用Java api运行
1)java工程-maven,依赖。
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>0.11.0.1</version> </dependency>
|
- 编写代码-写生产者的代码
/** * 订单的生产者代码 */ public class OrderProducer {
public static void main(String[] args) throws InterruptedException {
/* 1、连接集群,通过配置文件的方式 * 2、发送数据-topic:order,value */ Properties props = new Properties(); props.put("bootstrap.servers", "node01:9092"); props.put("acks", "all"); props.put("retries", 0); props.put("batch.size", 16384); props.put("linger.ms", 1); props.put("buffer.memory", 33554432); props.put("key.serializer",
"org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer",
"org.apache.kafka.common.serialization.StringSerializer"); KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(props);
for (int i = 0; i < 1000; i++) {
// 发送数据 ,需要一个producerRecord对象,最少参数 String topic, V value kafkaProducer.send(new ProducerRecord<String, String>("order", "订单信息!"+i)); Thread.sleep(100); } } }
|
- 编写代码-写消费者的代码
/** * 消费订单数据--- javaben.tojson */ public class OrderConsumer {
public static void main(String[] args) {
// 1\连接集群 Properties props = new Properties(); props.put("bootstrap.servers", "node01:9092"); props.put("group.id", "test"); props.put("enable.auto.commit", "true"); props.put("auto.commit.interval.ms", "1000"); props.put("key.deserializer",
"org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer",
"org.apache.kafka.common.serialization.StringDeserializer"); KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(props); // 2、发送数据 发送数据需要,订阅下要消费的topic。 order kafkaConsumer.subscribe(Arrays.asList("order"));
while (true) { ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(100);// jdk queue offer插入、poll获取元素。 blockingqueue put插入原生,take获取元素 for (ConsumerRecord<String, String> record : consumerRecords) { System.out.println("消费的数据为:" + record.value()); } } } }
|
6、Apache Kafka原理
6.1、Apache Kafka原理-分片与副本机制
bin/kafka-topics.sh --create --zookeeper zk01:2181 --replication-factor 1 --partitions 1--topic order
分片:solrcloud中有提及到。
当数据量非常大的时候,一个服务器存放不了,就将数据分成两个或者多个部分,存放在多台服务器上。每个服务器上的数据,叫做一个分片。
副本:solrcloud中有提及到。
当数据只保存一份的时候,有丢失的风险。为了更好的容错和容灾,将数据拷贝几份,保存到不同的机器上。
6.2、Apache Kafka原理-消息不丢失机制
6.2.1、生产者端消息不丢失
- 消息生产分为同步模式和异步模式
- 消息确认分为三个状态
- 0:生产者只负责发送数据
- 1:某个partition的leader收到数据给出响应
- -1:某个partition的所有副本都收到数据后给出响应
- 在同步模式下
- 生产者等待10S,如果broker没有给出ack响应,就认为失败。
- 生产者重试3次,如果还没有响应,就报错。
- 在异步模式下
- 先将数据保存在生产者端的buffer中。Buffer大小是2万条。
- 满足数据阈值或者数量阈值其中的一个条件就可以发送数据。
- 发送一批数据的大小是500条。
如果broker迟迟不给ack,而buffer又满了。
开发者可以设置是否直接清空buffer中的数据。
6.2.2、Borker端消息不丢失
broker端的消息不丢失,其实就是用partition副本机制来保证。
Producer ack -1.能够保证所有的副本都同步好了数据。其中一台机器挂了,并不影像数据的完整性。
6.2.3、消费者端消息不丢失
只要记录offset值,消费者端不会存在消息不丢失的可能。只会重复消费。
6.3、Apache Kafka原理-消息存储及查询机制
6.3.1、文件存储机制
segment段中有两个核心的文件一个是log,一个是index。 当log文件等于1G时,新的会写入到下一个segment中。
通过下图中的数据,可以看到一个segment段差不多会存储70万条数据。
6.3.2、文件查询机制
6.4、Apache Kafka原理-生产者数据分发策略
kafka在数据生产的时候,有一个数据分发策略。默认的情况使用DefaultPartitioner.class类。
这个类中就定义数据分发的策略。
- 如果是用户制定了partition,生产就不会调用DefaultPartitioner.partition()方法
- 当用户指定key,使用hash算法。如果key一直不变,同一个key算出来的hash值是个固定值。如果是固定值,这种hash取模就没有意义。
Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions
- 当用既没有指定partition也没有key。
数据分发策略的时候,可以指定数据发往哪个partition。
当ProducerRecord的构造参数中有partition的时候,就可以发送到对应partition上
/** * Creates a record to be sent to a specified topic and partition * * @param topic The topic the record will be appended to * @param partition The partition to which the record should be sent * @param key The key that will be included in the record * @param value The record contents */ public ProducerRecord(String topic, Integer partition, K key, V value) {
this(topic, partition, null, key, value, null); }
|
如果生产者没有指定partition,但是发送消息中有key,就key的hash值。
/** * Create a record to be sent to Kafka * * @param topic The topic the record will be appended to * @param key The key that will be included in the record * @param value The record contents */ public ProducerRecord(String topic, K key, V value) {
this(topic, null, null, key, value, null); }
|
既没有指定partition,也没有key的情况下如何发送数据。
使用轮询的方式发送数据。
/** * Create a record with no key * * @param topic The topic this record should be sent to * @param value The record contents */ public ProducerRecord(String topic, V value) {
this(topic, null, null, null, value, null); }
|
6.5、Apache Kafka原理-消费者的负载均衡机制
一个partition只能被一个组中的成员消费。
所以如果消费组中有多于partition数量的消费者,那么一定会有消费者无法消费数据。
7、了解Apache Kafka 监控及运维
7.1、一键启动Kafka
7.2 、UI界面