济南网站优化培训,桂林象鼻山作文400字,制作网站工具,在线制作视频的网站Kafka是一种高吞吐量的分布式发布订阅消息系统#xff0c;因为其高吞吐量、分布式可扩展性等等强大功能使得在目前互联网系统中广泛使用。该篇博客入门了解一下Kafka的安装及使用。
Kafka概念
Kafk是分布式消息队列。Kafka对消息保存时根据Topic进行归类#xff0c;发送消息…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.0Zookeeper 3.5.2Windows 7Java 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_90223、配置server.properties
broker.id三份都需要唯一目前设置为0,1,2
broker.id0配置服务器端口因为是单机所以IP地址一样需要端口不一样。分别设置9020、9021、9022
listenersPLAINTEXT://:9020设置log地址分别设置/kafka_9020/、/kafka_9021/、/kafka_9022/
log.dirsD:/kafka/KafkaCluster/kafka_9020/kafka-logs并添加配置可删除Topic如果不配置Kafka只是标记删除
delete.topic.enabletrue4、启动三个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\windowskafka-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\windowskafka-topics.bat --list --zookeeper localhost:2181
kafka-topic-test查看Topic详情
kafka-topics.bat --zookeeper [Zookeeper地址] --describe --topic [topic名称]测试
D:\kafka\KafkaCluster\kafka_9020\bin\windowskafka-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\windowskafka-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\windowskafka-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\windowskafka-console-producer.bat --broker-list localhost:9020 --topic kafka_topic_test
hello
world
kafka消费端9021
D:\kafka\KafkaCluster\kafka_9021\bin\windowskafka-console-consumer.bat --bootstrap-server localhost:9020 --from-beginning --topic kafka_topic_test
hello
world
kafka消费端9022
D:\kafka\KafkaCluster\kafka_9022\bin\windowskafka-console-consumer.bat --bootstrap-server localhost:9020 --from-beginning --topic kafka_topic_test
hello
world
kafkaJava操作Kafka
引入Jar
dependencygroupIdorg.apache.kafka/groupIdartifactIdkafka-clients/artifactIdversion2.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);//创建生产者对象KafkaProducerString,String producer new KafkaProducer(props);//循环发送消息for(int i10;i20;i){producer.send(new ProducerRecordString, 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)KafkaConsumerObject, Object consumer new KafkaConsumer(props);//指定Topic//consumer.subscribe(Arrays.asList(first,second,third));consumer.subscribe(Collections.singletonList(kafka-topic-test));while (true) {//获取数据ConsumerRecordsObject, Object consumerRecords consumer.poll(Duration.ofMillis(100));for (ConsumerRecordObject, Object consumerRecord : consumerRecords) {System.out.println(consumerRecord.topic():consumerRecord.partition():consumerRecord.value());}}}
}
