概述

Kafka是Apache软件基金会的开源的流处理平台，该平台提供了消息的订阅与发布，能够基于Kafka实现对网络日志流实时在线处理，在这个维度上弥补了Hadoop的离线分析系统的不足。因为基于hadoop的MapReduce系统分析离线数据延迟较高，而且不支持动态数据处理和分析。Kafka的流处理平台不仅仅可以为离线系统储备数据（通常使用Kafka作为数据缓冲），而且Kafka自身也提供了一套数据流的处理机制，实现对数据流在线处理，比如： 统计。

• 作为消息队列(Message Queue)充当系统的缓冲组件 - MiddleWare
• 作为一套在线实时流处理组件 (轻)

场景分析

• 异步消息
  使用Kafka MQ功能实现模块间异步通信，把一些费时的操作交给额外的服务或者设备去执行，这样可以提升系统运行效率，加速连接释放的速度，例如：用户注册模块，在用户注册成功后，业务系统需要给用户发送一个通知短信，通知用户登录邮箱去激活刚注册的用户信息。这种业务场景如图所示，因为短信通知和邮件发送是一个比较耗时的操作，所以在这里没必要将短信和邮件发送作为注册模块的流程，使用Message Queue功能可以将改业务和主业务注册分离，这样可以缩短用户浏览器和服务建立的链接时间，同时也能满足发送短信和邮件的业务。
  
• 系统间解耦|削峰填谷
  ①在某些高吞吐的业务场景下，可能会出现在某一个时间段系统负载写入的负载压力比较大，短时间有大量的数据需要持久化到数据库中，但是由于数据的持久化需要数据库提供服务，由于传统的数据库甚至一些NoSQL产品也不能很好的解决高并发写入，因为数据库除去要向用户提供链接之外，还需要对新来的数据做持久化，这就需要一定的时间才能将数据落地到磁盘。因此在高并发写入的场景，就需要用户集成Message Queue在数据库前作为缓冲队列。在队列的另一头只需要程序有条不紊的将数据写入到数据库即可，这就保证无论外界写入压力有多么大都可以借助于Message Queue缓解数据库的压力。
  
  ②Message Queue除了解决对数据缓冲的压力之外，还可以充当业务系统的中间件（Middleware）作为系统服务间解耦的组件存在，例如上图所示订单模块和库存模块中就可以使用Message Queue作为缓冲队列实现业务系统服务间的解耦，也就意味着即使服务在运行期间库存系统宕机也并不会影响订单系统的正常运行。

架构概念

集群

Kafka集群以Topic形式负责管理集群中的Record，每一个Record属于一个Topic。底层Kafka集群通过日志分区形式持久化Record。在Kafka集群中，Topic的每一个分区都一定会有1个Borker担当该分区的Leader，其他的Broker担当该分区的follower（取决于分区的副本因子）。一旦对应分区的Lead宕机，kafka集群会给当前的分区指定新的Borker作为该分区的Leader。分区的Leader的选举是通过Zookeeper一些特性实现的，这里就不在概述了。Leader负责对应分区的读写操作，Follower负责数据备份操作。

日志&分区

Kafka集群是通过日志形式存储Topic中的Record，Record会根据分区策略计算得到的分区数存储到相应分区的文件中。每个分区都是一个有序的，不可变的记录序列，不断附加到结构化的commit-log中。每个分区文件会为Record进去分区的顺序进行编排。每一个分区中的Record都有一个id，该id标示了该record进入分区的先后顺序，通常将该id称为record在分区中的offset偏移量从0开始，依次递增。

Kafka集群持久地保留所有已发布的记录 - 无论它们是否已被消耗 - 使用可配置的保留时间。例如，如果保留策略设置为2天，则在发布记录后的2天内，它可供使用，之后将被丢弃以释放空间。Kafka的性能在数据大小方面实际上是恒定的，因此长时间存储数据不是问题。
[外链图片转存失败(img-dIHmuWnJ-1562201509823)(http://kafka.apache.org/0110/images/log_consumer.png)]
事实上，基于每个消费者保留的唯一元数据是该消费者在日志中的偏移或位置。这种offset由消费者控制：通常消费者在读取记录时会线性地增加其偏移量，但事实上，由于消费者控制位置，它可以按照自己喜欢的任何顺序消费记录。例如，消费者可以重置为较旧的偏移量以重新处理过去的数据，或者跳到最近的记录并从“现在”开始消费。

分区数目决定系统对外的吞吐能力，分区数目越大吞吐性能越好。通常来说队列一定保证FIFO，但是由于Kafka采取了hash（key）%分区数的分区策略将数据发送到对应的分区中，因此Kafka的Topic只能保证分区内部数据遵循FIFO策略。

生产者

生产者负责发送Record到Kafka集群中的Topic中。在发布消息的时候，首先先计算Record分区计算方案有三种：①如果用户没有指定分区但是指定了key信息，生产者会根据hash（key）%分区数计算该Record所属分区信息，②如果生产者在发送消息的时候并没有key，也没有指定分区数，生产者会使用轮训策略选择分区信息。③如果指定了分区信息，就按照指定的分区信息选择对应的分区；当分区参数确定以后生产者会找到相应分区的Leader节点将Record记录写入到Topic日志存储分区中。

消费者

消费者作为消息的消费者，消费者对Topic中消息的消费是以Group为单位进行消费，Kafka服务会自动的按照组内和组间对消费者消费的分区进行协调。

• 组内均分分区，确保一个组内的消费者不可重复消费分区中的数据，一般来说一个组内的消费者实例对的数目应该小于或者等于分区数目。
• 组间广播形式消费，确保所有组都可以拿到当前Record。组间数据之间可以保证对数据的独立消费。

安装和部署

环境筹备

• 准备三台物理主机主机名分别是CentOSA|B|C

• 安装JDK，配置JAVA_HOME，本案例安装JDK1.8+

• 校准物理主机时钟，确保时间一致。

• 配置主机名和IP的映射关系，这是必须的，因为Kafka默认只认主机名，最后别忘记关闭防火墙。

• 安装Zookeeper集群，并且保证Zookeeper能正常运行。

• 下载Kafka服务安装包http://archive.apache.org/dist/kafka/2.2.0/kafka_2.11-2.2.0.tgz

安装配置

这里为了描述方便，如果是CentOSA|CentOSB|CentOSC三台物理主机都要配置的信息会使用CentOSX代替，表示该操作三台机器都要修改或者执行。

[root@CentOSX ~]# tar -zxf kafka_2.11-2.2.0.tgz -C /usr/
[root@CentOSX ~]# vi /usr/kafka_2.11-2.2.0/config/server.properties
############################# Server Basics #############################
broker.id=[0|1|2]
delete.topic.enable=true
############################# Socket Server Settings #############################
listeners=PLAINTEXT://CentOS[A|B|C]:9092
############################# Log Basics #############################
log.dirs=/usr/kafka-logs
############################# Log Retention Policy #############################
log.retention.hours=168
############################# Zookeeper #############################
zookeeper.connect=CentOSA:2181,CentOSB:2181,CentOSC:2181

本案例中安装的是kafka_2.11-2.2.0.tgz版本，由于Kafka底层使用的Scala和Java混编，因此在kafka发行版本例如：kafka_2.11-2.2.0.tgz其中2.11是Scala的编译版本，因为Scala兼容Java所以运行Kafka无需安装Scala环境；2.2.0是kafka的版本号。Kafka从0.11.x以后加入事务等特性的支持。

配置说明

配置项	说明
broker.id	每一台Kafka服务的id信息，必须设置不同。
delete.topic.enable	配置该属性开启删除topic的能力，否则kafka无法删除Topic信息。
listeners	配置Kafka服务的监听服务入口。
log.dirs	配置Kafka日志存储路径，存储消息信息。
log.retention.hours	日志存储时间，一旦日志数据超过改时间，系统会自动删除过期日志。
zookeeper.connect	zookeeper链接参数信息，用于保存Kafka元数据信息。

启动服务

[root@CentOSX ~]# cd /usr/kafka_2.11-2.2.0/
[root@CentOSX kafka_2.11-2.2.0]# ./bin/kafka-server-start.sh -daemon config/server.properties
[root@CentOSX kafka_2.11-2.2.0]# jps
5507 Kafka
2358 QuorumPeerMain
5660 Jps

关闭服务

kafka-server-stop.sh自带的服务脚本中存在的一些问题，用户需要修改该脚本文件，具体修改内容如下所示：

[root@CentOSX kafka_2.11-2.2.0]# vi bin/kafka-server-stop.sh

SIGNAL=${SIGNAL:-TERM}
PIDS=$(jps | grep  Kafka | awk '{print $1}')if [ -z "$PIDS" ]; then
echo "No kafka server to stop"
exit 1
else
kill -s $SIGNAL $PIDS
fi

Kafka自带脚本中PIDS参数获取存在问题，导致每次获取的PIDS都是空信息。

验证服务

• 创建topic

[root@CentOSA kafka_2.11-2.2.0]# ./bin/kafka-topics.sh
--zookeeper CentOSA:2181,CentOSB:2181,CentOSC:2181
--create
--topic topic01
--partitions 3
--replication-factor 3

partitions:日志分区数；replication-factor:分区副本因子

• 消费者

[root@CentOSA kafka_2.11-2.2.0]# ./bin/kafka-console-consumer.sh 
--bootstrap-server CentOSA:9092,CentOSB:9092,CentOSC:9092 
--topic topic01

• 生产者

[root@CentOSB kafka_2.11-2.2.0]# ./bin/kafka-console-producer.sh 
--broker-list CentOSA:9092,CentOSB:9092,CentOSC:9092 
--topic topic01
> hello kafka

观察CentOSA控制台输出，如果有hello kafka说明安装成功！

Topic管理篇（DDL）

• 创建Tocpic

[root@CentOSA kafka_2.11-2.2.0]# ./bin/kafka-topics.sh
--zookeeper CentOSA:2181,CentOSB:2181,CentOSC:2181
--create
--topic topic01
--partitions 3
--replication-factor 3

• Topic详细信息

[root@CentOSC kafka_2.11-2.2.0]# ./bin/kafka-topics.sh 
--describe 
--zookeeper CentOSA:2181,CentOSB:2181,CentOSC:2181 
--topic topic01
Topic:topic01   PartitionCount:3        ReplicationFactor:3     Configs:
Topic: topic01  Partition: 0    Leader: 2       Replicas: 1,2,0 Isr: 2,0,1
Topic: topic01  Partition: 1    Leader: 2       Replicas: 2,0,1 Isr: 2,0,1
Topic: topic01  Partition: 2    Leader: 2       Replicas: 0,1,2 Isr: 2,0,1

• 删除Topic

[root@CentOSA kafka_2.11-2.2.0]# ./bin/kafka-topics.sh 
--zookeeper CentOSA:2181,CentOSB:2181,CentOSC:2181 
--delete 
--topic topic01

如果用户没有配置delete.topic.enable=true，则Topic删除不起作用。

• Topic列表

[root@CentOSA kafka_2.11-2.2.0]# ./bin/kafka-topics.sh 
--zookeeper CentOSA:2181,CentOSB:2181,CentOSC:2181 
--list

Kafka API实战

maven

<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.2.0</version>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-streams</artifactId>
<version>2.2.0</version>
</dependency><dependency>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
<version>1.2.17</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
<version>1.7.5</version>
</dependency>

在Windos配置主机名和IP映射关系

192.168.40.129 CentOSA
192.168.40.130 CentOSB
192.168.40.131 CentOSC

必须配置主机名和IP的映射关系，否则运行主机在连接kafka服务的时候，会抛出无法解析主机异常或者链接超时，这一点是很多初学者在使用Kafka的时候容易忽略的一点。

log4j.properties

### set log levels ###
log4j.rootLogger = info,stdout 
### 输出到控制台 ###
log4j.appender.stdout = org.apache.log4j.ConsoleAppender
log4j.appender.stdout.Target = System.out
log4j.appender.stdout.layout = org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern =  %d{ABSOLUTE} %5p %c{1}:%L - %m%n

Topic管理

管理Topic的核心在于创建AdminClient，通过adminClient完成对Topic的基础管理

//创建AdminClient
Properties props = new Properties();
props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,"链接参数");
AdminClient adminClient=KafkaAdminClient.create(props);// todo your code here//关闭连接
adminClient.close();

Topic创建

List<NewTopic> newTopics = Arrays.asList(new NewTopic("topic01", 3, (short)2));
adminClient.createTopics(newTopics);

Topic列表

ListTopicsResult topics = adminClient.listTopics();
topics.names()
.get()
.stream()
.forEach((topic)-> System.out.println(topic));

Topic详情

adminClient.describeTopics(Arrays.asList("topic01")).all().get().entrySet().stream().forEach((entry)-> {String topic=entry.getKey();System.out.println(topic);TopicDescription descr = entry.getValue();List<TopicPartitionInfo> partitions = descr.partitions();for (TopicPartitionInfo partition : partitions) {System.out.println("\t"+partition);}});

Topic删除

adminClient.deleteTopics(Arrays.asList("topic01"));

集群状态

adminClient.describeCluster().nodes().get().stream().forEach((node)-> System.out.println(node) );

生产者

生产者负责产生消息，并且将生产的消息发送到kafka集群中，在Kaka集群中所有发送的消息都必须是以二进制分区日志形式存储，因此生产者在发送的数据之前需要指定数据序列化规则，在创建生产者之前的基本步骤如下：

Properties props=new Properties();
//配置Kafka生产者 ...
Producer<String,String> producer=new KafkaProducer<String, String>(props);producer.close();

这里需要了解生产者常见的属性配置及其含义。

属性	默认值	含义	是否必须
bootstrap.servers	“”	连接kafka集群连接参数	必须
key.serializer	null	key序列化规则	必须
value.serializer	null	value序列化规则	必须
acks	1	生产者要求leader在考虑完成请求之前收到的确认数量。	否
retries	2147483647	当没有在规定时间内acker，则认定发送失败，重试	否
batch.size	16384	一次缓冲多少数据，并不是一条数据就会触发发送	否
linger.ms	0	间隔多长时间构建一次batch发送	否
request.timeout.ms	30000	设置客户端最大等待超时时间	否
enable.idempotence	false	是否开启幂等性，可以保证生产者一个Record只发送一次给broker	否

public static Producer<String,String> buildKafkaProducer(){
Properties props=new Properties();
//必须配置属性
props.put("bootstrap.servers","CentOSA:9092,CentOSB:9092,CentOSC:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");//优化参数 选配
props.put("acks","all");
props.put("retries",1);
props.put("batch.size",1024);
props.put("linger.ms",5);
props.put("request.timeout.ms",10000);
props.put("enable.idempotence",true);return new KafkaProducer<String, String>(props);
}

发送一则消息

public static void main(String[] args) {
Producer<String,String> producer=buildKafkaProducer();ProducerRecord<String,String> record=
new ProducerRecord<String, String>("topic01","001","TV,GAME,PRGRAMING");
producer.send(record);
//强制刷新缓冲区,防止虚拟机退出数据还没来得及发送出去
producer.flush();
producer.close();
}

消费者

消费者负责消费集群中的消息，消费者消费Topic中的消息是按照group消费形式订阅的。所以在创建消费者的时候和生产者类似也需要按照如下步骤：

Properties props = new Properties();
//配置消费者 ...
Consumer<String,String> consumer=new KafkaConsumer<String, String>(props);
consumer.close();

这里需要了解消费者常见的属性配置及其含义。

属性	默认值	含义	是否必须
bootstrap.servers	“”	连接kafka服务器参数	必须
key.deserializer	null	key反序列化	必须
value.deserializer	null	value反序列化	必须
group.id	“”	如果是订阅方式，必须指定组id	必须
enable.auto.commit	true	offset自动提交	否
auto.commit.interval.ms	5000	自动提交频率	否

public static Consumer<String, String> buildKafkaConsumer() {
Properties props = new Properties();
props.put("bootstrap.servers", "CentOSA:9092,CentOSB:9092,CentOSC:9092");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
props.put("group.id", "g1");
props.put("enable.auto.commit", true);
props.put("auto.commit.interval.ms", 5000);
return new KafkaConsumer<String, String>(props);
}

消费消息

Consumer<String,String> consumer=buildKafkaConsumer();List<String> topics = Arrays.asList("topic01");
consumer.subscribe(topics);while (true){
//间隔5秒钟查询一次，看服务器中有没有Record
ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(5));
}

偏移量管理

默认消费端对分区数据的偏移量提交方式是自动的通过两个参数限定：

enable.auto.commit=true
auto.commit.interval.ms=5000

有可能导致消费端并没有正常消费数据但是，客户端会自动提交偏移量。因此为了解决该问题用户一般可以考虑手动提交偏移量。

1)现关闭offset自动提交

enable.auto.commit=false

2）用户在消费完数据之后手动提交偏移量

Consumer<String,String> consumer=buildKafkaConsumer();List<String> topics = Arrays.asList("topic01");
consumer.subscribe(topics);
while (true){
//间隔1秒钟查询一次，看服务器中有没有Record
ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(5));
Map<TopicPartition,OffsetAndMetadata> commits= new HashMap<TopicPartition, OffsetAndMetadata>();
for (ConsumerRecord<String, String> record : records) {
// to do your code here//表示下一次消费偏移量起始位置，因此要在当前位置+1
OffsetAndMetadata offsetAndMetadata = new OffsetAndMetadata(record.offset()+1);
//构建分区信息
TopicPartition partition = new TopicPartition(record.topic(), record.partition());
//存储需要提交分区参数
commits.put(partition,offsetAndMetadata);
}
//提交offset
consumer.commitSync(commits);

事务控制

单独生产者

transactional.id=transaction-1
enable.idempotence=true

Producer<String,String> producer=buildKafkaProducer();
//1.初始化事务
producer.initTransactions();
producer.beginTransaction();
try {
for(int i=10;i<20;i++){
DecimalFormat decimalFormat=new DecimalFormat("000");
ProducerRecord<String,String> record=new ProducerRecord<String, String>("topic01",
decimalFormat.format(i),
"TV,GAME,PRGRAMING");
producer.send(record);
}
//事务提交
producer.commitTransaction();
} catch (Exception e) {
//终止事务
producer.abortTransaction();
}
//强制刷新缓冲区
producer.flush();
producer.close();

消费者事务隔离级别

isolation.level=read_committed

Consumer<String,String> consumer=buildKafkaConsumer();//订阅感兴趣的Topic
List<String> topics = Arrays.asList("topic01");
consumer.subscribe(topics);
//定期的去查询队列里有没有关注的Record
while (true){
//间隔1秒钟查询一次，看服务器中有没有Record
ConsumerRecords<String, String> records = consumer.poll(1000);
Map<TopicPartition,OffsetAndMetadata> commits=new HashMap<TopicPartition, OffsetAndMetadata>();
for (ConsumerRecord<String, String> record : records) {
System.out.println("key:"+record.key()+",value:"+record.value()+
" ,ts:"+record.timestamp()+"\tp|o:\t"+record.partition()+"|"+record.offset());
TopicPartition partition = new TopicPartition(record.topic(), record.partition());
OffsetAndMetadata offsetAndMetadata = new OffsetAndMetadata(record.offset()+1);
commits.put(partition,offsetAndMetadata);
}
consumer.commitSync(commits);
}

Consumer -> Producer 事务控制

public static void main(String[] args) {
Consumer<String, String> consumer = buildKafkaConsumer();
Producer<String,String> producer=buildKafkaProducer();
//初始化事务
producer.initTransactions();
consumer.subscribe(Arrays.asList("topic01"));
while (true){
//开启事务控制
producer.beginTransaction();
ConsumerRecords<String, String> records = consumer.poll(1000);
try {
Map<TopicPartition,OffsetAndMetadata> commits=new HashMap<TopicPartition, OffsetAndMetadata>();
for (ConsumerRecord<String, String> record : records) {
//封装数据offset
TopicPartition partition = new TopicPartition(record.topic(), record.partition());
OffsetAndMetadata offsetAndMetadata = new OffsetAndMetadata(record.offset()+1);
commits.put(partition,offsetAndMetadata);
// 做一些额外逻辑 ...//将计算后的信息写入到下一个Topic中
ProducerRecord<String, String> srecord = new ProducerRecord<String, String>("topic02", record.key(), record.value());
producer.send(srecord);
//int n=1/0;
}
producer.flush();
//提交指定消费组的偏移量
producer.sendOffsetsToTransaction(commits,"g1");
producer.commitTransaction();
} catch (Exception e) {
producer.abortTransaction();
}
}
}
public static Producer<String,String> buildKafkaProducer(){
Properties props=new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"CentOSA:9092,CentOSB:9092,CentOSC:9092");
//配置 Kafka key、value序列化
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
//设置ACKS应答,等待所有的Broker应答
props.put(ProducerConfig.ACKS_CONFIG,"all");
//设置及回话超时时间
props.put(ProducerConfig.RETRIES_CONFIG,1);
//设置一次发送多少个Record
props.put(ProducerConfig.BATCH_SIZE_CONFIG,1024);
//设置一个延迟发送记录时间 延迟5ms发送
props.put(ProducerConfig.LINGER_MS_CONFIG,5);
//设置响应超时时间
props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG,10000);
//开启Kafka的幂等写
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,true);
//开启事务控制
props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG,"transaction-2");return new KafkaProducer<String, String>(props);
}
public static Consumer<String,String> buildKafkaConsumer(){
Properties props=new Properties();//配置Broker服务器的地址
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"CentOSA:9092,CentOSB:9092,CentOSC:9092");
//设置数据反序列化
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
//设置消费者所属组信息
props.put(ConsumerConfig.GROUP_ID_CONFIG,"g1");
//设置消费者offset控制
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);
//设置offset偏移量提交频率时间间隔 5s
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,5000);
props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG,"read_committed");return new KafkaConsumer<String, String>(props);
}

更多精彩内容关注