2 消息队列的分类消息是什么？任何可以传递的数据都是消息。无论是视频、音频、文字等等的数据都是消息。
队列是什么？数据结构，先进先出。
消息+队列：保留消息的队列。消息在传输过程中，保留消息的一种容器(队列)。
作用：
主要提供生产、消费接口供外部调用做数据存储和获取。
2.1 P2PMQ分为两类：P2P（Peer 2 Peer） Pub/Sub(发布/订阅)
2.2 Pub/Sub 2.3 常见的消息中间件一般基于poll和push。
3 Kafka概述RabbitMQ Erlang编写，同时支持p2p和pub/sub
Redis ， 他是支持MQ功能的，可以做轻量级的消息队列
ActiveMQ， p2p
Kafka/Jafka ， 高性能的分布式的消息中间件
RocketMQ, 纯Java开发的。
3.1 Kafka的关键概念 Kafka是一个分布式的基于发布 与订阅的消息中间件。它最初是有领英公司发布，是由scala语言编写的。后来被纳入到apache的顶级项目。
4 安装Kafkatopic : 主题，kafka生产和消费消息的分类。
broker ： 消息代理服务器，kafka集群中的一个kafka的服务器节点都是broker。主要作用存消息。消息是以主题为单位存储，一个主题可以分为多个分区。
partition ：Topic在物理上的分组，一个topic在一个broker被分为一个或者多个partition。分区在创建主题的时候就要指定。
message ： 消息，kafka通信的基本单位。每个消息都存储在具体的partition中。
producer ： 生产者，向kafka的topic发送message
consumer ：消费者，从kafka的topic拉取消息
zookeeper ： 协调kafka服务
##1. 解压并配置环境变量 [root@hadoop software]# tar -zxvf kafka_2.12-2.4.1.tgz -C /opt/apps/ & cd /opt/apps/ [root@hadoop apps]# mv kafka_2.12-2.4.1/ kafka-2.4.1/ & cd kafka-2.4.1/ [root@hadoop kafka-2.4.1]# vi /etc/profile ## KAFKA export KAFKA_HOME=/opt/apps/kafka-2.4.1 export PATH=$PATH:$JAVA_HOME/bin:$SCALA_HOME/bin:$HADOOP_HOME/bin:$HADOOP_HOME/sbin:$HIVE_HOME/bin:$SPARK_HOME/bin:$SPARK_HOME/sbin: $KAFKA_HOME/bin ##2. server.properties [root@hadoop kafka-2.4.1]# vi config/server.properties ############################# Server Basics ############################# # The id of the broker. This must be set to a unique integer for each broker. broker.id=1 ############################# Socket Server Settings ############################# # 如果你要使用生产者和消费者来访问你的kafka，你又是一个云服务器，就必须要配置。（同学不用配置） advertised.listeners=PLAINTEXT://10.206.0.4:9092 ############################# Log Basics ############################# # kafka日志记录的路径，如果设置多个路径可以使用逗号分割 log.dirs=/opt/apps/kafka-2.4.1/logs ############################# Zookeeper ############################# zookeeper.connect=hadoop:2181/kafka ##3. zookeeper.properties(zoo.cfg) ##4. 如果你搭建的是全分布式，请你将server.properties拷贝给其他两台机器（注意broker.id不能相同） ##5. 如果你搭建的是全分布式，需要你自己安装zookeeper的集群，然后启动kafka集群的时候，必须要先启动zk的集群。（每个节点都要启动一次） ##6. 伪分布式启动kafka集群 ##6.1 先启动kafka自带的zk的服务 [root@hadoop bin]# zookeeper-server-start.sh -daemon $KAFKA_HOME/config/zookeeper.properties ##6.2 连接zk [root@hadoop bin]# zookeeper-shell.sh localhost:2181 ls / [zookeeper] ##6.3 启动kafka [root@hadoop bin]# kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties5 Kafka的Topic的命令 5.1 创建主题
##1. 这是过时的 [root@hadoop ~]# kafka-topics.sh --create > --topic spark > --zookeeper hadoop:2181/kafka > --partitions 3 > --replication-factor 1 Created topic spark. ##2. 不过时的 [root@hadoop ~]# kafka-topics.sh --create > --topic hadoop > --bootstrap-server hadoop:9092 > --partitions 3 > --replication-factor 15.2 例举所有的主题
[root@hadoop ~]# kafka-topics.sh --list > --bootstrap-server hadoop:9092 hadoop spark5.3 查询主题详情
[root@hadoop ~]# kafka-topics.sh --describe
> --topic hadoop
> --bootstrap-server hadoop:9092
Topic: hadoop PartitionCount: 3 ReplicationFactor: 3 Configs: segment.bytes=1073741824
Topic: hadoop Partition: 0 Leader: 2 Replicas: 2,3,1 Isr: 2,3,1
Topic: hadoop Partition: 1 Leader: 3 Replicas: 3,1,2 Isr: 3,1,2
Topic: hadoop Partition: 2 Leader: 1 Replicas: 1,2,3 Isr: 1,2,3
Topic : 当前主题名称
PartitionCount ： 分区总数
ReplicationFactor ： 副本因子总数
segment.bytes ： 日志存储文件的字节大小
Replicas ： 副本因子，当前kafka对应的partition所在的broker.id的列表
Leader ：该partition的所有副本中的leader，负责处理该partition的读写
Isr ： 该partition存活的副本对应的broker.id
Topic: hadoop Partition: 0 Leader: 1 Replicas: 1 Isr: 1
5.4 修改主题
##1. 修改hadoop主题，将3个分区添加成4个分区 [root@hadoop ~]# kafka-topics.sh --alter > --topic hadoop > --partitions 4 > --bootstrap-server hadoop:9092 ## tip:分区数修改的时候只能增不能减少5.5 删除主题
[root@hadoop ~]# kafka-topics.sh --delete > --topic hadoop > --bootstrap-server hadoop:9092 [root@hadoop ~]# kafka-topics.sh --list > --bootstrap-server hadoop:9092 spark6 命令行方式的生产和消费 6.1 生产者
[root@hadoop bin]# kafka-console-producer.sh > --topic spark > --broker-list hadoop:90926.2 消费者
[root@hadoop bin]# kafka-console-consumer.sh > --topic spark > --bootstrap-server hadoop:90927 Kafka在Zookeeper中的目录说明
##1. 集群id目录
/kafka
/cluster
/id
get /kafka/cluster/id
{"version":"1","id":"ge98jALkRjqociNOapucfw"}
##tip : 我们通过以上命令可以找看到集群的id
##2. controller
/kafka
/controller
get /kafka/controller
{"version":1,"brokerid":1,"timestamp":"1654050431732"}
##tip: 在本集群中的broker.id=1的broker它是一个controller
## 在kafka集群中controller起到了非常只管紧要的作用，控制partition中的副本的leader的选举、控制topic的CRUD。一个集群中只有一个controller。
##3. controller_epoch
/kafka
/controller_epoch
get /kafka/controller_epoch
1
##tip: 1表示kafka集群中的controller的纪元，说白了就是你的controller的迭代。
##4. brokers
/kafka
/brokers
ls /kafka/brokers
[ids, seqid, topics]
## ids : 有哪些id
ls /kafka/brokers/ids
[1，2，3]
## topics ： 主题
ls /kafka/brokers/topics
[__consumer_offsets, spark]
## seqid ： 系统的序列id
##5. consumers
##tip:在早期版本的kafka中适用于专门存储消费者偏移量的目录，在新版本的kafka中已经被弃用了。新版本的kafka将消费者的偏移量保存在了一个系统主题中：__consumer_offsets
8 Kafka的消费者消费并行度的决定因素
9 关于消费者偏移量 二 Kafka API 1 导入kafka依赖 Kafka在进行消费的时候，消费者是可以分组的(我们称之为消费者组)，一个消费者组中可以有多个消费者对象。然后Kafka的一个主题下可以有多个分区，主题是逻辑上的概念，真实的数据是存储在主题的分区下的，由于我们一个主题下有多个分区，我们在消费的时候消费的是一个主题，如何才能够最大化消费我们的这个主题，手段就是设置消费者组中的消费者个数恰好和主题的分区数一样。
所以，结论。消费的并行度是由主题的分区数以及消费者组中的消费者个数决定的。
2 Topic增删改查 2.1 创建主题4.0.0 com.qf.bigdata kafka 1.0-SNAPSHOT 2.4.1 1.8 1.8 org.apache.kafka kafka_2.12 ${kafka-version}
package com.qf.bigdata.kafka.day1;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.NewTopic;
import java.io.IOException;
import java.util.Arrays;
import java.util.Properties;
public class Demo3_Admin {
public static void main(String[] args) throws IOException {
//1. 获取到操作kafka的admin对象
Properties properties = new Properties();
properties.load(Demo3_Admin.class.getClassLoader().getResourceAsStream("admin.properties"));
AdminClient adminClient = AdminClient.create(properties);
//2. 创建主题
adminClient.createTopics(Arrays.asList(new NewTopic("hadoop", 3, (short)1)));
//3. 释放资源
adminClient.close();
}
}
2.2 列举所有的主题
package com.qf.bigdata.kafka.day1;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.KafkaFuture;
import java.io.IOException;
import java.util.Arrays;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ExecutionException;
public class Demo3_Admin {
public static void main(String[] args) throws IOException, ExecutionException, InterruptedException {
//1. 获取到操作kafka的admin对象
Properties properties = new Properties();
properties.load(Demo3_Admin.class.getClassLoader().getResourceAsStream("admin.properties"));
AdminClient adminClient = AdminClient.create(properties);
//2. 列举主题
ListTopicsResult result = adminClient.listTopics();
//2.1 获取到结果中的所有的主题的名字
KafkaFuture> listTopic = result.names();
//2.2 获取到具体的名称集合
Set names = listTopic.get();
//2.3 遍历
for(String name : names) {
System.out.println(name);
}
//3. 释放资源
adminClient.close();
}
}
2.3 查询主题详情
package com.qf.bigdata.kafka.day1;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.TopicPartitionInfo;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.ExecutionException;
public class Demo3_Admin {
public static void main(String[] args) throws IOException, ExecutionException, InterruptedException {
//1. 获取到操作kafka的admin对象
Properties properties = new Properties();
properties.load(Demo3_Admin.class.getClassLoader().getResourceAsStream("admin.properties"));
AdminClient adminClient = AdminClient.create(properties);
//2. 获取到spark主题的详情
DescribeTopicsResult result = adminClient.describeTopics(Arrays.asList("spark"));
//2.1 获取到result中的具体的参数
KafkaFuture> all = result.all();
//2.2 获取到具体的数值
Map map = all.get(); //
//2.3 遍历
for (Map.Entry entry : map.entrySet()) {
System.out.println(entry.getKey());
System.out.println(entry.getValue());
System.out.println(entry.getValue().name()); // topicName
List partitions = entry.getValue().partitions();
for (TopicPartitionInfo info : partitions) {
System.out.println(info.partition() + "->" + info.leader() + "->" + info.replicas() + "->" + info.isr());
}
}
//3. 释放资源
adminClient.close();
}
}
3 生产者API
package com.qf.bigdata.kafka.day1;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.io.IOException;
import java.util.Properties;
public class Demo1_Producer {
public static void main(String[] args) throws IOException {
//1. 创建配置对象
Properties properties = new Properties();
properties.load(Demo1_Producer.class.getClassLoader().getResourceAsStream("producer.properties"));
//2. 创建一个生产者对象
Producer producer = new KafkaProducer(properties);
//3. 发送消息
ProducerRecord record = new ProducerRecord("spark", "lixi");
producer.send(record);
//4. 释放资源
producer.close();
}
}
4 消费者API
package com.qf.bigdata.kafka.day1;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.io.IOException;
import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;
public class Demo2_Consumer {
public static void main(String[] args) throws IOException {
//1. 创建配置文件
Properties properties = new Properties();
properties.load(Demo2_Consumer.class.getClassLoader().getResourceAsStream("consumer.properties"));
//2. 创建消费者对象
Consumer consumer = new KafkaConsumer(properties);
//3. 订阅主题
consumer.subscribe(Arrays.asList("spark"));
//4. 拉取消息
boolean loop = true;
while (loop) {
//4.1 获取到一条记录，1000毫秒拉取一次
ConsumerRecords records = consumer.poll(Duration.ofMillis(1000));
//4.2 获取到消息的详情
for (ConsumerRecord record : records) {
Integer key = record.key();
String value = record.value();
int partition = record.partition();
String topic = record.topic();
long offset = record.offset();
System.out.println(key + " " + value + " " + partition + " " + topic + " " + offset);
}
}
consumer.close();
}
}
5 分区
5.1 默认的kafka的分区策略
5.2 自定义分区器 5.2.1 Partitioner 每条producerRecord有topic名称、partition编号，并且我们的记录实际是key/value的数据。我们的分区策略是：
- 如果生产者在发送消息的时候有指定分区，那么这条消息就是直接给指定的主题以及我们自己指定的分区发送消息。
- 如果我们生产的时候没有指定的分区，但是指定了key值，使用key的hash来选择分区。
- 如果既没有指定分区，又没有指定key，那么默认使用轮询的方式分配分区。
public interface Partitioner extends Configurable, Closeable {
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster);
public void close();
default public void onNewBatch(String topic, Cluster cluster, int prevPartition) {
}
}
5.2.2 Configurable
public interface Configurable {
void configure(Map configs);
}
5.2.3 随机分区器
- RandomPartitioner
package com.qf.bigdata.kafka.day2;
import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import java.util.Map;
import java.util.Random;
public class RandomPartitioner implements Partitioner {
private Random random = new Random();
@Override
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
//1. 获取到当前主题一共有多少个分区
Integer partitionCount = cluster.partitionCountForTopic(topic);
//2. 生成随机数
int partitionId = random.nextInt(partitionCount);
return partitionId;
}
@Override
public void close() {
}
@Override
public void configure(Map configs) {
}
}
- 注册分区器——producer.properties
partitioner.class=com.qf.bigdata.kafka.day2.RandomPartitioner三 Flume和Kafka整合 1 介绍 2 安装Flume
[root@hadoop software]# tar -zxvf apache-flume-1.9.0-bin.tar.gz -C /opt/apps/ & cd /opt/apps/ [root@hadoop apps]# mv apache-flume-1.9.0-bin/ flume-1.9.0 [root@hadoop conf]# mv flume-env.sh.template flume-env.sh export JAVA_HOME=/opt/apps/jdk1.8.0_261 ## FLUME export FLUME_HOME=/opt/apps/flume-1.9.0 export PATH=$PATH:$FLUME_HOME/bin [root@hadoop ~]# flume-ng version Flume 1.9.0 Source code repository: https://git-wip-us.apache.org/repos/asf/flume.git Revision: d4fcab4f501d41597bc616921329a4339f73585e Compiled by fszabo on Mon Dec 17 20:45:25 CET 2018 From source with checksum 35db629a3bda49d23e9b3690c80737f93 创建kafka主题
[root@hadoop logs]# kafka-topics.sh --create --topic flume-kafka --partitions 3 --replication-factor 1 --bootstrap-server hadoop:90924 创建flume的配置文件 4.0 下载telnet
[root@hadoop ~]# yum -y install telnet [root@hadoop ~]# telnet hadoop 100864.1 flume-kafka-sink.conf
[root@hadoop conf]# vi flume-kafka-sink.conf a1.sources = r1 a1.channels = c1 a1.sinks = k1 # SOURCE a1.sources.r1.type = netcat a1.sources.r1.bind = hadoop a1.sources.r1.port = 10086 # CHANNEL a1.channels.c1.type = memory a1.channels.c1.capacity = 10000 a1.channels.c1.transactionCapacity = 10000 # SINK a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink a1.sinks.k1.kafka.topic = flume-kafka a1.sinks.k1.kafka.bootstrap.servers = hadoop:9092 a1.sources.r1.channels = c1 a1.sinks.k1.channel = c14.2 开启消费者端
[root@hadoop conf]# kafka-console-consumer.sh --topic flume-kafka --bootstrap-server hadoop:9092 --from-beginning4.3 启动flume
##1. 前台启动 flume-ng agent -n a1 -c /opt/apps/flume-1.9.0/conf -f /opt/apps/flume-1.9.0/conf/flume-kafka-sink.conf -Dflume.root.logger=INFO,console ##2. 后台启动 nohup flume-ng agent -n a1 -c /opt/apps/flume-1.9.0/conf -f /opt/apps/flume-1.9.0/conf/flume-kafka-sink.conf > /dev/null 2>&1 &
