Kafka介绍使用

一、Kafka介绍

Apache Kafka是一个发布/订阅的消息系统，在2011年开源。Kafka具有很多特性：

1. Kafka本身支持分布式，很容易横向扩展

2. 高吞吐量，高性能

3. 高容错性

kafka术语：

• Broker：Kafka的服务器端由被称为Broker的服务进程构成，即一个Kafka集群由多个Broker组成，Broker负责接收和处理客户端发送过来的请求，以及对消息进行持久化。

• 备份机制：一个leader，多个follower。只有leader提供服务，leader向follower同步数据，follower只做一件事，向leader发送请求，请求leader把最新消息发给它以保证主从同步。

• Partition：分区，将每个主题划分为多个分区，每一个分区是一组有序的消息日志。每条消息只会被发送到一个分区中，上面的leader和follower就是对分区说的，一个分区有一个leader和多个follower。

• 持久化：使用消息日志持久化，就是一个只能追加写的物理文件，通过日志段（log segment）回收磁盘，一个日志细分为多个日志段，消息总是被写入最新的日志段，当写满一个日志段后，kafka会自动切出一个新的日志段，kafka后台有定时任务定时判断老的日志段是否要删除。

• 消息位移：offset，表示分区中每条消息的位置信息，是一个单调递增且不变的值。

• 消费者位移：Consumer Offset。表征消费者消费进度，每个消费者都有自己的消费者位移。

• 消费者组：Consumer Group。多个消费者实例共同组成的一个组，同时消费多个分区以实现高吞吐。

• 重平衡：Rebalance。消费者组内某个消费者实例挂掉后，其他消费者实例自动重新分配订阅主题分区的过程。Rebalance是Kafka消费者端实现高可用的重要手段。

二、安装启动kafka

# 下载kafka
wget http://mirror.bit.edu.cn/apache/kafka/2.4.0/kafka_2.13-2.4.0.tgz
# 解压
tar -zxvf kafka_2.13-2.4.0.tgz
cd kafka_2.13-2.4.0
# 目录如下
bin：包含Kafka运行的所有脚本，如：start/stop Zookeeper，start/stop Kafka
libs：Kafka运行的依赖库
config：zookeeper，Logger，Kafka等相关配置文件
sit-docs：Kafka相关文档

kafka能够以三种方式部署集群：

• 单节点-单Broker集群：只在一个节点上部署一个Broker

• 单节点-多Broker集群：在一个节点上部署多个Broker，只不过各个Broker以不同的端口启动

• 多节点-多Broker集群：以上两种的组合，每个节点上部署一到多个Broker，且各个节点连接起来

kafka使用Zookeeper来存储集群元数据和Consumer信息，因此有两种选项：

• 独立部署Zookeeper应用并部署kafka集群到该Zookeeper；

• 使用kafka自带的Zookeeper；

为了方便测试，下面使用kafka自带的Zookeeper来启动。

# 启动Zookeeper，默认的端口号是2181，可以在./config/zookeeper.properties修改，本次测试已改为2182
./bin/zookeeper-server-start.sh ./config/zookeeper.properties
# 启动kafka
./bin/kafka-server-start.sh ./config/server.properties 
......
[2020-02-21 15:49:10,389] INFO Kafka commitId: 77a89fcf8d7fa018 (org.apache.kafka.common.utils.AppInfoParser)
[2020-02-21 15:49:10,389] INFO Kafka startTimeMs: 1582271350379 (org.apache.kafka.common.utils.AppInfoParser)
[2020-02-21 15:49:10,393] INFO [KafkaServer id=0] started (kafka.server.KafkaServer)

三、使用kafka

# 创建名为topic1的topic
./bin/kafka-topics.sh --create --zookeeper localhost:2182 --replication-factor 1 --partitions 1 --topic topic1
# 列出所有的topic
 ./bin/kafka-topics.sh --list --zookeeper localhost:2182
# 启动一个Producer并发送消息
./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic topic1
OpenJDK Client VM warning: G1 GC is disabled in this release.
>hello
>kafka
>
# 启动另一个Consumer并接受消息
./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topic1 --from-beginning
OpenJDK Client VM warning: G1 GC is disabled in this release.
hello
kafka

四、使用java api

依赖kafka的client包：

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

生产者源码：

Properties properties = new Properties();
        properties.put("bootstrap.servers", "127.0.0.1:9092");
        properties.put("acks", "all");
        properties.put("retries", 0);
        properties.put("batch.size", 16384);
        properties.put("linger.ms", 1);
        properties.put("buffer.memory", 33554432);
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        Producer<String, String> producer = null;
        try {
            producer = new KafkaProducer<String, String>(properties);
            for (int i = 0; i < 100; i++) {
                String msg = "Message " + i;
                producer.send(new ProducerRecord<String, String>("topic1", msg));
                System.out.println("Sent:" + msg);
            }
        } catch (Exception e) {
            e.printStackTrace();

        } finally {
            producer.close();
        }

五、分区策略

所谓分区策略就是生产者决定将消息发送到哪个分区的算法。

• 轮询策略：就是顺序分配，是kafka的默认策略；

• 随机策略：就是随机将消息发到某个分区上；

• 按消息键保序策略：kafka允许为每条消息定义消息键，一旦消息被定义了消息键，那么kafka就保证同一个消息键的消息会发送到同一个分区上，由于分区上的消息是有序的，所以叫消息键保序策略，如果消息定义了消息键，这个就是默认策略，否则默认策略是轮询。

• 自定义策略：你需要显式地配置生产者端的参数partitioner.class。然后在编写生产者程序时，你可以编写一个具体的类实现org.apache.kafka.clients.producer.Partitioner接口。

六、kafka的一些能力

• 拦截器：生产者拦截器支持在消息发送前以及消息提交成功后植入拦截器逻辑，消费者拦截器支持在消息消费前以及提交位移后植入拦截器逻辑。可用于监控，审计，性能检测等场景。

• 消息可靠性：kafka提供了三种消息生产可靠性保证：至多一次，至少一次，精确一次。精确一次通过幂等性Producer和事务性Producer实现，其中幂等性保证单会话单分区的精确一次的消息生产，多分区和多会话不保证，比如重启后就不能保证。而事务性Producer则能保证多分区和多会话的精确一次的消息生产。

参考：

http://www.itrensheng.com/archives/apache-kafka-java-api