Kafka安装与使用

Kafka是一种高吞吐量的分布式发布订阅消息系统，因为其高吞吐量、分布式可扩展性等等强大功能使得在目前互联网系统中广泛使用。该篇博客入门了解一下Kafka的安装及使用。

Kafka概念

Kafk是分布式消息队列。Kafka对消息保存时根据Topic进行归类，发送消息者称为Producer，消息接收者称为Consumer。此外kafka集群有多个kafka实例组成，每个实例(server)称为broker。其中每个Topic都由若干个partition组成，partition是topic物理上的分组，每个partition是一个有序的队列。Kafka的消费端有位移(offset)的概念，每条消息在某个partition的位移是固定的，相当于在分区当中的唯一编号。无论是kafka集群，还是consumer都依赖于Zookeeper集群保存一些meta信息，来保证系统可用性。

Kafka集群配置

因为本机是Windows系统，测试方便就单机配置Kafka的集群，但是配置都是共通的，在服务器上也基本一样。

配置版本

• Kafka 2.2.0
• Zookeeper 3.5.2
• Windows 7
• Java 8

配置Zookeeper

1、官网根据版本下载Zookeeper

2、解压Zookeeper的下载包，修改zoo.cfg中的dataDir地址，也可修改端口

3、点击zkServer.cmd，启动Zookeeper

配置Kafka

1、官网根据版本下载Kafka

2、解压Kafka的下载包，并复制三份，用于配置集群

本机的目录

D:\kafka\KafkaCluster\kafka_9020
D:\kafka\KafkaCluster\kafka_9021
D:\kafka\KafkaCluster\kafka_9022

3、配置server.properties

broker.id三份都需要唯一，目前设置为0,1,2

broker.id=0

配置服务器端口，因为是单机所以IP地址一样，需要端口不一样。分别设置9020、9021、9022

listeners=PLAINTEXT://:9020

设置log地址，分别设置/kafka_9020/、/kafka_9021/、/kafka_9022/

log.dirs=D:/kafka/KafkaCluster/kafka_9020/kafka-logs

并添加配置可删除Topic，如果不配置，Kafka只是标记删除

delete.topic.enable=true

4、启动三个Kafka服务器

分别在主目录/kafka_9020/、/kafka_9021/、/kafka_9022/主目录CMD窗口运行

.\bin\windows\kafka-server-start.bat .\config\server.properties

上述正常即可配置成功。

测试Kafka

配置是否成功，我们可以使用命令行操作查看。本机是Windows所以使用的都是bat文件，若到Linux则用sh文件。

创建Topic

kafka-topics.bat --create --zookeeper [Zookeeper地址] --partitions [分区数] --replication-factor [副本集数] --topic [topic名称]

注意，副本集数不能大于不能大于Broker数，这里Broker数为3

测试

D:\kafka\KafkaCluster\kafka_9020\bin\windows>kafka-topics.bat --create --zookeeper localhost:2181 --partitions 1 
--replication-factor 2 --topic kafka-topic-testCreated topic kafka_topic_test.

查看Topic列表

kafka-topics.bat --list --zookeeper [Zookeeper地址]

测试

D:\kafka\KafkaCluster\kafka_9020\bin\windows>kafka-topics.bat --list --zookeeper localhost:2181
kafka-topic-test

查看Topic详情

kafka-topics.bat --zookeeper [Zookeeper地址] --describe --topic [topic名称]

测试

D:\kafka\KafkaCluster\kafka_9020\bin\windows>kafka-topics.bat --zookeeper localh
ost:2181 --describe --topic kafka_topic_test
Topic:kafka_topic_test  PartitionCount:1        ReplicationFactor:2     Configs:
Topic:kafka_topic_test  Partition: 0    Leader: 1       Replicas: 1,2  Isr: 1,2

删除Topic

kafka-topics.bat --delete --zookeeper [Zookeeper地址] --topic [topic名称]

测试

D:\kafka\KafkaCluster\kafka_9020\bin\windows>kafka-topics.bat --delete --zookeeper localhost:2181 --topic kafka_topic_test
Topic kafka_topic_test is marked for deletion.
Note: This will have no impact if delete.topic.enable is not set to true.D:\kafka\KafkaCluster\kafka_9020\bin\windows>kafka-topics.bat --list --zookeeper localhost:2181
__consumer_offsets
kafka_topic_test - marked for deletion

消息生产和消费

启动生产端

kafka-console-producer.bat --broker-list [broker地址] --topic [topic名称]

启动消费端

kafka-console-consumer.bat --zookeeper [Zookeeper地址] --from-beginning --topic [topic名称]

测试

生产端9020

D:\kafka\KafkaCluster\kafka_9020\bin\windows>kafka-console-producer.bat --broker-list localhost:9020 --topic kafka_topic_test
>hello
>world
>kafka
>

消费端9021

D:\kafka\KafkaCluster\kafka_9021\bin\windows>kafka-console-consumer.bat --bootstrap-server  localhost:9020 --from-beginning --topic kafka_topic_test
hello
world
kafka

消费端9022

D:\kafka\KafkaCluster\kafka_9022\bin\windows>kafka-console-consumer.bat --bootstrap-server  localhost:9020 --from-beginning --topic kafka_topic_test
hello
world
kafka

Java操作Kafka

引入Jar

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

消费端

import java.util.Properties;import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;public class CustomerProducer {public static void main(String[] args) {//http://kafka.apache.org/documentation/#producerconfigs 更多配置可以访问此地址//配置信息Properties props = new Properties();//设置kafka集群的地址 -- localhost:9020,localhost:9021,localhost:9022props.put("bootstrap.servers", "localhost:9020,localhost:9021,localhost:9022");//ack模式，all是最慢但最安全的// 0  不等待成功返回  // 1  等Leader写成功返回  //all 等Leader和所有ISR中的Follower写成功返回,all也可以用-1代替props.put("acks", "all");//失败重试次数props.put("retries", 0);//每个分区未发送消息总字节大小（单位：字节），超过设置的值就会提交数据到服务端props.put("batch.size", 16384);//请求的最大字节数，该值要比batch.size大//不建议去更改这个值，如果设置不好会导致程序不报错，但消息又没有发送成功//props.put("max.request.size",1048576);//消息在缓冲区保留的时间，超过设置的值就会被提交到服务端//数据在缓冲区中保留的时长,0表示立即发送//为了减少网络耗时，需要设置这个值，太大可能容易导致缓冲区满，阻塞消费者，太小容易频繁请求服务端props.put("linger.ms", 1);//整个Producer用到总内存的大小，如果缓冲区满了会提交数据到服务端//buffer.memory要大于batch.size，否则会报申请内存不足的错误//不要超过物理内存，根据实际情况调整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> producer = new KafkaProducer<>(props);//循环发送消息for(int i=10;i<20;i++){producer.send(new ProducerRecord<String, String>("kafka-topic-test", Integer.toString(i)),new Callback() {@Overridepublic void onCompletion(RecordMetadata metadata, Exception exception) {if(exception == null){System.out.println(metadata.partition()+" - "+metadata.offset());}else{System.out.println("发送失败");}}});}//关闭资源producer.close();}}

生产端

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;public class CustomerConsumer {public static void main(String[] args) {Properties props = new Properties();//设置kafka集群的地址props.put("bootstrap.servers", "localhost:9020,localhost:9021,localhost:9022");//消费者组IDprops.put("group.id", "test-consumer-group");//设置自动提交offsetprops.put("enable.auto.commit", "true");//自动提交间隔props.put("auto.commit.interval.ms", "1000");//earliest //当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费//latest//当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据//none//topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常//默认建议用earliest。设置该参数后 kafka出错后重启，找到未消费的offset可以继续消费。props.put("auto.offset.reset", "earliest");//Consumer session 过期时间props.put("session.timeout.ms", "30000");//反序列化器props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");//创建消费者对象@SuppressWarnings("resource")KafkaConsumer<Object, Object> consumer = new KafkaConsumer<>(props);//指定Topic//consumer.subscribe(Arrays.asList("first","second","third"));consumer.subscribe(Collections.singletonList("kafka-topic-test"));while (true) {//获取数据ConsumerRecords<Object, Object> consumerRecords = consumer.poll(Duration.ofMillis(100));for (ConsumerRecord<Object, Object> consumerRecord : consumerRecords) {System.out.println(consumerRecord.topic()+":"+consumerRecord.partition()+":"+consumerRecord.value());}}}
}