kafka基本知识概述

一、概述

1、什么是MQ

消息中间件，通过消息的发送和消息的接收分离实现应用程序的解耦，但是这个是MQ的效果并非目的，真正的目的是为了通讯，屏蔽底层复杂的通讯协议，定义一套更加简单的通讯，一个分布式系统中两个模块之间通讯要么是HTTP，要么是自己开发的（rpc） TCP，但是这两种协议其实都是原始的协议。HTTP 协议很难实现两端通讯——模块 A 可以调用 B，B 也可以主动调用 A，如果要做到这个两端都要背上WebServer，而且还不支持⻓连接（HTTP 2.0 的库根本找不到）。TCP 就更加原始了，粘包、心跳、私有的协议，想一想头皮就发麻。MQ 所要做的就是在这些协议之上构建一个简单的“协议”——生产者/消费者模型。MQ 带给我的“协议”不是具体的通讯协议，而是更高层次通讯模型。它定义了两个对象——发送数据的叫生产者；接收数据的叫消费者， 提供一个SDK 让我们可以定义自己的生产者和消费者实现消息通讯而无视底层通讯协议

2、MQ分类

2.1、有broker

又分为中topic和轻topic

kafka、JMS（ActiveMQ）就属于这个流派，生产者会发送 key 和数据到 Broker，由 Broker比较 key 之后决定给哪个消费者。这种模式是我们最常⻅的模式，是我们对 MQ 最多的印象。在这种模式下一个 topic 往往是一个比较大的概念，甚至一个系统中就可能只有一个topic，topic 某种意义上就是 queue，生产者发送 key 相当于说：“hi，把数据放到 key 的队列中”

2.2、无broker

这种的代表是 RabbitMQ（或者说是 AMQP）。生产者发送 key 和数据，消费者定义订阅的队列，Broker 收到数据之后会通过一定的逻辑计算出 key 对应的队列，然后把数据交给队列

3、kafka

基于分布式的分布/订阅消息队列，主要用于数据实时处理领域

官网：Apache Kafka

4、消息队列的两种模式

点对点模式，一对一，消费者主动拉取数据，消息收到后消息清除

消息生产者生产消息发送到Queue中，然后消息消费者从Queue中取出并且消费消息。消息被消费以后， queue 中不再有存储，所以消息消费者不可能消费到已经被消费的消息。Queue 支持存在多个消费者，但是对一个消息而言，只会有一个消费者可以消费。

发布订阅模式，一对多，消费者消费数据之后不会清除消息

消息生产者（发布）将消息发布到 topic 中，同时有多个消息消费者（订阅）消费该消息。和点对点方式不同，发布到 topic 的消息会被所有订阅者消费。

二、基本架构和概念

1. Producer ： 消息生产者，就是向 Kafka发消息 ；
2. Consumer ： 消息消费者，向 Kafka 取消息的客户端；
3. Consumer Group （CG）： 消费者组，由多个 consumer 组成。 消费者组内每个消费者负责消费不同分区的数据，一个分区只能由一个组内消费者消费；消费者组之间互不影响。 所有的消费者都属于某个消费者组，即消费者组是逻辑上的一个订阅者。
4. Broker ：经纪人 一台 Kafka 服务器就是一个 broker。一个集群由多个 broker 组成。一个 broker可以容纳多个 topic。
5. Topic ： 话题，可以理解为一个队列， 生产者和消费者面向的都是一个 topic；
6. Partition： 为了实现扩展性，一个非常大的 topic 可以分布到多个 broker（即服务器）上，一个 topic 可以分为多个 partition，每个 partition 是一个有序的队列；
7. Replica： 副本（Replication），为保证集群中的某个节点发生故障时， 该节点上的 partition 数据不丢失，且 Kafka仍然能够继续工作， Kafka 提供了副本机制，一个 topic 的每个分区都有若干个副本，一个 leader 和若干个 follower。
8. Leader： 每个分区多个副本的“主”，生产者发送数据的对象，以及消费者消费数据的对象都是 leader。
9. Follower： 每个分区多个副本中的“从”，实时从 leader 中同步数据，保持和 leader 数据的同步。 leader 发生故障时，某个 Follower 会成为新的 leader。

三、安装

1、window上安装

kafka(3版本之后不需要安装zk)

在Windows系统上安装消息队列kafka

2、Linux安装

2.1、安装

需要提前安装好java和zk

下载kafka的压缩包:http://kafka.apache.org/downloads

解压到路径 /usr/local/kafka

修改配置文件/usr/local/kafka/kafka2.11-2.4/config/server.properties

#broker.id属性在kafka集群中必须要是唯一
broker.id= 0
#kafka部署的机器ip和提供服务的端口号
listeners=PLAINTEXT://192.168.65.60:9092
#kafka的消息存储文件
log.dir=/usr/local/data/kafka-logs
#kafka连接zookeeper的地址
zookeeper.connect= 192.168.65.60:2181

启动kafka

./kafka-server-start.sh -daemon../config/server.properties

验证是否启动成功：

进入到zk中的节点看id是 0 的broker有没有存在（上线）

ls /brokers/ids/

2.2、server.properties核心配置详解：

显示详细信息

四、基本知识

1、创建主题、生产消息、消费消息

执行以下命令创建名为“test”的topic，这个topic只有一个partition，并且备份因子也设置为1

./kafka-topics.sh --create --zookeeper 172.16.253.35:2181 --replication-factor 1 --partitions 1 --topic test

查看当前kafka内有哪些topic

./kafka-topics.sh --list --zookeeper 172.16.253.35:2181

kafka自带了一个producer命令客户端，可以从本地文件中读取内容，或者我们也可以以命令行中直接输入内容，并将这些内容以消息的形式发送到kafka集群中。在默认情况下，每一个行会被当做成一个独立的消息。使用kafka的发送消息的客户端，指定发送到的kafka服务器地址和topic

./kafka-console-producer.sh --broker-list 172.16.253.38:9092 --topic test

对于consumer，kafka同样也携带了一个命令行客户端，会将获取到内容在命令中进行输 出， 默认是消费最新的消息 。使用kafka的消费者消息的客户端，从指定kafka服务器的指定 topic中消费消息

方式一：从最后一条消息的偏移量+1开始消费

./kafka-console-consumer.sh --bootstrap-server 172.16.253.38:9092 --topic testCopy to clipboardErrorCopied

方式二：从头开始消费

./kafka-console-consumer.sh --bootstrap-server 172.16.253.38:9092 --from-beginning --topic test

2、单播消息、多播消息

单播场景：同一个消费组里的多个消费者只有一个消费者能消费到某一个topic消息

多播场景：一个topic中的一条消息要被多个消费者消费，需要让不同的消费者处于不同的消费组

3、消费组详细信息

# 查看当前主题下有哪些消费组

./kafka-consumer-groups.sh --bootstrap-server 10.31.167.10:9092 --list

# 查看消费组中的具体信息：比如当前偏移量、最后一条消息的偏移量、堆积的消息数量

./kafka-consumer-groups.sh --bootstrap-server 172.16.253.38:9092 --describe --group testGroup

● current-offset:最后被消费的消息的偏移量

● Log-end-offset:消息总量（最后一条消息的偏移量）

● Lag:积压了多少条消息

4、小结

• 消息是被存储的，发送给broker保存至本地的日志文件中(配置文件中配置的目录)
• 消息的保存是有顺序的，通过offset偏移量来描述消息的有序性
• 消费者消费消息时也是通过offset描述当前要消费的那条消息的位置

五、主题、分区、_comsumer_offsets

一个topic中的文件可以非常多，多到几个g，存在log文件中，为了解决这个文件问题，引入了partition分区，分开存储一个topic中的消息，比如一个topic三个partition，，这个topic中的消息可以分三个地方部署，partition不仅解决了文件存储过大的问题，同时支持读写多个分区

给主题创建多个分区

./kafka-topics.sh --create --zookeeper localhost:2181 --partitions 2 --topic test1

查看topic主题的分区信息

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

实际上是存在data/kafka-logs/test1-0 和 test1-1中的0000000.log文件中

000.index:相当于一个稀疏索引文件

000.timeindex：相当于一个时间索引文件

000.log就是消息

kafka内部自己创建了一个_consumer_offsets主题，这个主题有49个分区，在data/log目录下可以查看，kafka会定期将分区的offset提交给这个主题，即这个主题主要用来保存消费者消费消息的偏移量，比如一个test主题有100条消息，一个消费者两给消费者a和b,a从1消费到50突然挂掉了，此时kafka会将偏移量50保存到_consumer_offsets主题中，b消费的时候从51消费

保存到_consumer_offsets时，key是consumerGroupId+topic+分区号，value就是当前的offset，

因为要处理高并发，所以默认有50个分区，可以通过offset.num.topic.partitions设置，这样通过加机器抗大并发，通过如下公式计算要提交到_consumer_offset中的哪个分区

公式：hash(consumerGroupId)%_consumer_offsets分区数

文件中保存的消息，默认保存7天，7天到期自动清除

六、kafka集群、副本

1、搭建

略

2、副本

在一个集群即3个broker，同一个主题创建2个分区 3个人副本

./kafka-topics.sh --create --zookeeper 172.16.253.35:2181 --replication-factor 3 --partitions 2 --topic my-replicated-topic

查看详情

./kafka-topics.sh --describe --zookeeper 172.16.253.35:2181 --topic my-replicated-topic

kafka集群中由多个broker组成

一个broker中存放一个topic的不同partition——副本

leader：副本里的概念

• 每个partition都有一个broker作为leader。
• 消息发送方要把消息发给哪个broker？就看副本的leader是在哪个broker上面。副本里的leader专⻔用来接收消息。
• 接收到消息，其他follower通过poll的方式来同步数据。

通过kill掉leader之后查看

# kill掉leader

ps -aux | grep server.properties

kill 17631

# 查看topic情况

./kafka-topics.sh --describe --zookeeper 172.16.253.35:2181 --topic my-replicated-topic

follower：leader处理所有针对这个partition的读写请求，而follower被动复制leader，不提供读写（主要是为了保证多副本数据与消费的一致性），如果leader所在的broker挂掉，那么就会进行新leader的选举，至于怎么选，在之后的controller的概念中介绍。

isr： 可以同步的broker节点和已同步的broker节点，存放在isr集合中。

3、消息的生产和发送

#生产

./kafka-console-producer.sh --broker-list 172.16.253.38:9092,172.16.253.38:9093,172.16.253.38:9094 --topic my-replicated-topic

#消费

./kafka-console-consumer.sh --bootstrap-server 172.16.253.38:9092,172.16.253.38:9093,172.16.253.38:9094 --from-beginning --topic my-replicated-topic

• 一个partitions只能被一个消费组里的一个消费者消费目的是为了保证消费的顺序性，但是多个partition的多个消费者消费的消息顺序性是没法保证的，如何保证了？后续
• partition的数量决定了消费者的数量，建议同一个消费组里的消费者数量不要大于partition数量，多的消费者消费不到消息
• 如果消费者挂了，会触发rebalance机制，让其他消费者消费该分区

七、kafka java生产者和消费者

1、引入依赖

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.4.1</version>
</dependency>

2、生产消息

#### //消息的发送方
public class MyProducer {
 
private final static String TOPIC_NAME = "my-replicated-topic";
 
public static void main(String[] args) throws ExecutionException,InterruptedException {
Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"10.31.167.10:9092,10.31.167.10:9093,10.31.167.10:9094");
//把发送的key从字符串序列化为字节数组
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
//把发送消息value从字符串序列化为字节数组
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
 
Producer<String, String> producer = new KafkaProducer<String,String>(props);
 
Order order = new Order((long) i, i);
ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(TOPIC_NAME, order.getOrderId().toString(), JSON.toJSONString(order));
RecordMetadata metadata = producer.send(producerRecord).get();
//=====阻塞=======
System.out.println("同步方式发送消息结果：" + "topic-" +metadata.topic() + "|partition-"+ metadata.partition() + "|offset-" +metadata.offset());

发送到指定分区

ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(TOPIC_NAME, 0 , order.getOrderId().toString(), JSON.toJSONString(order));

未指定分区 根据业务key计算分区

//未指定发送分区，具体发送的分区计算公式：hash(key)%partitionNum
ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(TOPIC_NAME, order.getOrderId().toString(), JSON.toJSONString(order));

同步发消息

在收到kafka的ack告知发送成功之前一直处于阻塞状态

//等待消息发送成功的同步阻塞方法
RecordMetadata metadata = producer.send(producerRecord).get();
System.out.println("同步方式发送消息结果：" + "topic-" +metadata.topic() + "|partition-"+ metadata.partition() + "|offset-" +metadata.offset());

异步发送，发送完不用等boker回复，直接执行下面的业务逻辑，可以提供callback，让broker异步调用callback，告诉消息发送的结果

//要发送 5 条消息
Order order = new Order((long) i, i);
//指定发送分区
ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(TOPIC_NAME, 0 , order.getOrderId().toString(),JSON.toJSONString(order));
//异步回调方式发送消息
producer.send(producerRecord, new Callback() {
public void onCompletion(RecordMetadata metadata, Exception exception) {
if (exception != null) {
    System.err.println("发送消息失败：" +
    exception.getStackTrace());
}
if (metadata != null) {
System.out.println("异步方式发送消息结果：" + "topic-" +metadata.topic() + "|partition-"+ metadata.partition() + "|offset-" + metadata.offset());
         }
    }
});

ack参数配置，在同步调用情况下，消息发送时ack有三种配置

• （ 1 ）acks=0： 表示producer不需要等待任何broker确认收到消息的回复，就可以继续发送下一条消息。性能最高，但是最容易丢消息。
• （ 2 ）acks=1： 至少要等待leader已经成功将数据写入本地log，但是不需要等待所有follower是否成功写入。就可以继续发送下一条消息。这种情况下，如果follower没有成功备份数据，而此时leader又挂掉，则消息会丢失。
• （ 3 ）acks=-1或all： 需要等待 min.insync.replicas(默认为 1 ，推荐配置大于等于2) 这个参数配置的副本个数都成功写入日志，这种策略会保证只要有一个备份存活就不会丢失数据。这是最强的数据保证。一般除非是金融级别，或跟钱打交道的场景才会使用这种配置。

props.put(ProducerConfig.ACKS_CONFIG, "1");

其他

• 发送会默认会重试 3 次，每次间隔100ms
• 发送的消息会先进入到本地缓冲区（32mb），kakfa会跑一个线程，该线程去缓冲区中取16k的数据，发送到kafka，如果到 10 毫秒数据没取满16k，也会发送一次。

3、消费消息

3.1、基本实现

public class MyConsumer {
private final static String TOPIC_NAME = "my-replicated-topic";
private final static String CONSUMER_GROUP_NAME = "testGroup";
 
public static void main(String[] args) {
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"10.31.167.10:9092,10.31.167.10:9093,10.31.167.10:9094");
// 消费分组名
props.put(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP_NAME);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
//创建一个消费者的客户端
KafkaConsumer<String, String> consumer = new KafkaConsumer<String,String>(props);
// 消费者订阅主题列表
consumer.subscribe(Arrays.asList(TOPIC_NAME));
 
while (true) {
/*
* poll() API 是拉取消息的⻓轮询
*/
ConsumerRecords<String, String> records =consumer.poll(Duration.ofMillis( 1000 ));
for (ConsumerRecord<String, String> record : records) {
System.out.printf("收到消息：partition = %d,offset = %d, key =%s, value = %s%n", record.partition(),record.offset(), record.key(), record.value());
            }
        }
    }
}

3.2、设置自动提交offset——默认

// 是否自动提交offset，默认就是true
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
// 自动提交offset的间隔时间
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");

消费者poll到消息后默认情况下，会自动向broker的_consumer_offsets主题提交当前主题-分区消费的偏移量。

自动提交会丢消息： 因为如果消费者还没消费完poll下来的消息就自动提交了偏移量，那么此 时消费者挂了，于是下一个消费者会从已提交的offset的下一个位置开始消费消息。之前未被消费的消息就丢失掉了。

3.3、手动提交offset

设置手动提交参数

props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

消费完消息进行手动提交

手动同步提交

if (records.count() > 0 ) {
// 手动同步提交offset，当前线程会阻塞直到offset提交成功
// 一般使用同步提交，因为提交之后一般也没有什么逻辑代码了
consumer.commitSync();
}

手动异步提交

if (records.count() > 0 ) {
// 手动异步提交offset，当前线程提交offset不会阻塞，可以继续处理后面的程序逻辑
consumer.commitAsync(new OffsetCommitCallback() {
@Override
public void onComplete(Map<TopicPartition, OffsetAndMetadata>offsets, Exception exception) {
          if (exception != null) {
              System.err.println("Commit failed for " + offsets);
              System.err.println("Commit failed exception: " +exception.getStackTrace());
          }
       }
  });
}

3.4、消费者poll消息过程

• 消费者建立了与broker之间的⻓连接，开始poll消息。
• 默认一次poll 500条消息

props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500 );

可以根据消费速度的快慢来设置，因为如果两次poll的时间如果超出了30s的时间间隔，kafka会认为其消费能力过弱，将其踢出消费组。将分区分配给其他消费者。

可以通过这个值进行设置：

props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 30 * 1000 );

如果每隔1s内没有poll到任何消息，则继续去poll消息，循环往复，直到poll到消息。如果超出了1s，则此次⻓轮询结束。

ConsumerRecords<String, String> records =consumer.poll(Duration.ofMillis( 1000 ));

消费者发送心跳的时间间隔

props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 1000 );

kafka如果超过 10 秒没有收到消费者的心跳，则会把消费者踢出消费组，进行rebalance，把分区分配给其他消费者。

props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10 * 1000 );

3.5、指定分区消费

consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0 )));

3.6、消息回溯消费

consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0 )));
consumer.seekToBeginning(Arrays.asList(new TopicPartition(TOPIC_NAME,0 )));

3.7、指定offset消费

consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0 )));
consumer.seek(new TopicPartition(TOPIC_NAME, 0 ), 10 );

3.8、从指定时间点消费

List<PartitionInfo> topicPartitions =consumer.partitionsFor(TOPIC_NAME);
//从 1 小时前开始消费
long fetchDataTime = new Date().getTime() - 1000 * 60 * 60 ;
Map<TopicPartition, Long> map = new HashMap<>();
for (PartitionInfo par : topicPartitions) {
    map.put(new TopicPartition(TOPIC_NAME, par.partition()),fetchDataTime);
}
Map<TopicPartition, OffsetAndTimestamp> parMap =consumer.offsetsForTimes(map);
for (Map.Entry<TopicPartition, OffsetAndTimestamp> entry :parMap.entrySet()) {
    TopicPartition key = entry.getKey();
    OffsetAndTimestamp value = entry.getValue();
    if (key == null || value == null) continue;
    Long offset = value.offset();
    System.out.println("partition-" + key.partition() +"|offset-" + offset);
    System.out.println();
    //根据消费里的timestamp确定offset
    if (value != null) {
        consumer.assign(Arrays.asList(key));
        consumer.seek(key, offset);
    }
}

3.9、新消费组的消费偏移量

当消费主题的是一个新的消费组，或者指定offset的消费方式，offset不存在，那么应该如何消费?

• latest(默认) ：只消费自己启动之后发送到主题的消息
• earliest：第一次从头开始消费，以后按照消费offset记录继续消费，这个需要区别于consumer.seekToBeginning(每次都从头开始消费)

props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");