首页/小白筆記

实践 | kafka 基本使用

 2023-09-18 阅读 17 评论 0

摘要：Kafak概述kafka用于构建实时数据管道的应用中常被用到，可以横向扩展，容错，可以快速的运行在数以千计的公司的产品。可以像消息系统一样读写数据流，可以可靠的处理流相关的应用，可以安全的使用，即，kafka为一个消息中间件。场

Kafak概述

kafka用于构建实时数据管道的应用中常被用到，可以横向扩展，容错，可以快速的运行在数以千计的公司的产品。可以像消息系统一样读写数据流，可以可靠的处理流相关的应用，可以安全的使用，即，kafka为一个消息中间件。

场景比喻

我。生产者 你，消费者 馒头。数据流，消息 正常情况下，生产一个，消费一个。其他情况：
1）一直生产，你吃到某一个馒头时，你卡主(机器故障)， 馒头就丢失了 2）一直生产，做馒头速度快，你吃来不及，馒头也就丢失了 为了放着其他生产情况的出现，我们可以拿个碗/篮子，馒头做好以后先放到篮子里，你要吃的时候去篮子里面取出来吃，而这篮子/框就可以为：Kafka。当篮子满了，馒头就装不下了，咋办？多准备几个篮子 === Kafka的扩容

架构与概念

其中 producer：生产者，就是生产馒头 consumer：消费者，就是吃馒头的(你) broker：篮子 topic：主题，给馒头带一个标签，topica的馒头是给你吃的，topicb的馒头是给你弟弟吃

kafka快速入门，即 1.Kafka可以作为集群运行在一台或者多个服务器上面；

2.Kafka集群可以分类地存储记录流，以打标签的方式，就是采用topics，每个broker可以打个topic，这样能保证消费者可以根据topic选择性消费；

3.每个记录由Key、Value、timestamp构成。

快速使用

这里使用到了zookeeper

下载Kafka

打开链接https://www.apache.org/dyn/closer.cgi?path=/kafka/2.4.0/kafka_2.12-2.4.0.tgz

kafka应用，下载kafka的tgz安装包，

然后解压并进入kafka路径下，

> tar -xzf kafka_2.12-2.4.0.tgz
> cd kafka_2.12-2.4.0

启动服务

使用Kafka之前需要先启动一个ZooKeeper服务，这里直接使用Kafka中包含的脚本即可， 因为在前面下载的压缩包中已经包含了zookeeper相关的安装包。

> bin/zookeeper-server-start.sh config/zookeeper.properties

启动ZooKeeper服务之后再启动Kafka服务，

> bin/kafka-server-start.sh config/server.properties

需要强调一下，config/server.properties是Kafka的配置文件，可以用于配置监听的host、port、broker等。

slf4j kafka、默认的ZooKeeper连接服务为localhost:2181，

# Zookeeper connection string (see zookeeper docs for details).
# This is a comma separated host:port pairs, each corresponding to a zk
# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
# You can also append an optional chroot string to the urls to specify the
# root directory for all kafka znodes.
zookeeper.connect=localhost:2181

另外，producer和consumer的监听端口为9092，如果需要更改server的host和port端口可以通过修改config/server.properties进行配置。

基本使用

命令行

命令行的使用方式相对简单，通过前面的Kafka配置之后可以直接在命令行下进行使用。

创建Topic

使用Kafka，我们首先需要创建一个Topic，这样后续消息生产者和消息消费者才能针对性的发送和消费数据，

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

这样我们就创建了一个名为test的Topic。

发送消息

kafka实战。启动一个终端A，执行下面命令

> bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
>hello world
>

当执行producer脚本后，会出现消息输入提示符，这是我们可以输入消息(数据)，然后它会发送到对应的服务器(Broker)。

消费消息

现在管道中已经有了数据，接下来我就可以使用消费者去读取数据。

另外启动一个终端B，执行下面命令，

> ./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
hello world

Java

项目使用spring-boot和和spring-kafka，pom文件如下：

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.jc.demo</groupId><artifactId>jc-demo-kafka</artifactId><version>0.0.1.0-SNAPSHOT</version><packaging>jar</packaging><name>oneapp-archetype-test</name><url>http://www.jevoncode.com</url><properties><!-- Every File in Project Enconding --><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding><!-- Compiling Time Enconding --><maven.compiler.encoding>UTF-8</maven.compiler.encoding><!-- Compiling Time JDK Version --><java.version>1.7</java.version><!-- Test --><junit.version>4.12</junit.version><!-- Logging --><slf4j.version>1.7.21</slf4j.version><logback.version>1.1.7</logback.version></properties><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.0.3.RELEASE</version><relativePath/> <!-- lookup parent from repository --></parent><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter</artifactId></dependency><dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><!-- http://kafka.apache.org/documentation.html#producerapi--><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>1.1.0</version></dependency><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>${junit.version}</version><scope>test</scope></dependency><!-- Log依赖 --><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-api</artifactId><version>${slf4j.version}</version></dependency><!-- logback --><dependency><groupId>ch.qos.logback</groupId><artifactId>logback-classic</artifactId><version>${logback.version}</version></dependency><dependency><groupId>ch.qos.logback</groupId><artifactId>logback-core</artifactId><version>${logback.version}</version></dependency></dependencies></project>

配置文件

kafka与hadoop的区别、application.properties

#kafka地址
jc.kaHost=s1.jevoncode.com:9092,s2.jevoncode.com:9092,s3.jevoncode.com:9092

Topic配置

package com.jc.demo.springboot.config;import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.core.KafkaAdmin;import java.util.HashMap;
import java.util.Map;/*** 创建kafka的topic* 如果kafka不存在此topic则会自动创建，存在则不改变kafka的topic*/
@Configuration
@EnableKafka
public class TopicConfig {public static final String TOPIC_JC_KAFKA_DEMO = "jc-demo-kafka";@Value("${jc.kaHost}")String kaHost;@Beanpublic KafkaAdmin kafkaAdmin() {Map<String, Object> configs = new HashMap<>();configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kaHost);return new KafkaAdmin(configs);}@Beanpublic NewTopic foo() {//第一个是参数是topic名字，第二个参数是分区个数，第三个是topic的复制因子个数//当broker个数为1个时会创建topic失败，//提示：replication factor: 2 larger than available brokers: 1//只有在集群中才能使用kafka的备份功能//以kafka的分区机制来说，我将其numParitions个数设置为broker个数，也就是3return new NewTopic(TOPIC_JC_KAFKA_DEMO, 3, (short) 2);}//
//    @Bean
//    public NewTopic topic1(){
//        return new NewTopic("jc-demo-kafka2", 10, (short) 2);
//    }
}

启动类

package com.jc.demo.springboot;import com.jc.demo.springboot.config.ApplicationConfig;
import com.jc.demo.springboot.service.DoService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Import;
import org.springframework.context.annotation.PropertySource;@SpringBootApplication
@Import({ApplicationConfig.class})
@PropertySource("classpath:application.properties")
public class DemoApplication implements CommandLineRunner{@Autowiredprivate DoService doService;public static void main(String[] args) {SpringApplication.run(DemoApplication.class, args);}@Overridepublic void run(String... args) throws Exception {doService.HelloWorld();}
}

生产者

配置类KafkaProducerConfig

package com.jc.demo.springboot.config;import com.jc.demo.springboot.service.MyListener;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.*;import java.util.HashMap;
import java.util.Map;@Configuration
@EnableKafka
public class KafkaProducerConfig {@Value("${jc.kaHost}")String kaHost;/* --------------producer configuration-----------------**/@Beanpublic Map<String, Object> producerConfigs() {Map<String, Object> props = new HashMap<>();props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kaHost);props.put(ProducerConfig.RETRIES_CONFIG, 0);props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);props.put(ProducerConfig.LINGER_MS_CONFIG, 1);props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);return props;}@Beanpublic ProducerFactory<String, String> producerFactory() {return new DefaultKafkaProducerFactory<>(producerConfigs());}/* --------------kafka template configuration-----------------**/@Beanpublic KafkaTemplate<String, String> kafkaTemplate() {KafkaTemplate<String, String> kafkaTemplate = new KafkaTemplate<>(producerFactory());return kafkaTemplate;}}

服务接口和类（生产发送消息）

DoService接口

package com.jc.demo.springboot.service;public interface DoService {void HelloWorld();
}

DoService实现类，调用kafkaTemplate完成发送消息到kafka

package com.jc.demo.springboot.service;import com.jc.demo.springboot.config.TopicConfig;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;import java.math.BigInteger;@Service
public class DoServiceImpl implements  DoService {@AutowiredKafkaTemplate kafkaTemplate;@Overridepublic void HelloWorld() {String phone = "18689206965";while (true) {try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}kafkaTemplate.send(TopicConfig.TOPIC_JC_KAFKA_DEMO, partition(phone), phone, "jevoncode" + System.currentTimeMillis());}}/*** 根据手机号计算分区* @return*/private int partition(String phone) {int hash = phone.hashCode();int partition = new BigInteger(Integer.toString(hash)).mod(new BigInteger("3")).intValue();     //由于总共有3个分区，所以得去3的模System.out.println(partition);return partition;}}

此时执行DemoApplication的main方法，就可以生产一个字符串"jevoncode"到kafka的0分区上。可以使用命令查看：

[jevoncode@s1 kafka_2.11-1.1.0]$ ./bin/kafka-console-consumer.sh --bootstrap-server s1.jevoncode.com:9092 --topic jc-demo-kafka --from-beginning
jevoncode1531654603522
jevoncode1531654689283
jevoncode1531654690331
jevoncode1531654691332
jevoncode1531654692332
....

消费者

此时application.properties配置文件需增加消费者配置

#kafka地址
jc.kaHost=s1.jevoncode.com:9092,s2.jevoncode.com:9092,s3.jevoncode.com:9092#############以下是消费者端的配置###########################
#kafka消费者 groupId配置
jc.consumer.group.id=jc-consumer-group-1
#kafka消费者 分区配置，这样就可以指定每个消费者所消费的分区，提高吞吐量
jc.consumer.partitions=0,1,2
#一次从kafka拉的最大消息数
jc.max.poll.records=100

消费者配置类

package com.jc.demo.springboot.listener;import com.jc.demo.springboot.config.TopicConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;import java.util.List;public class KafkaListener {private static Logger logger = LoggerFactory.getLogger(KafkaListener.class);@org.springframework.kafka.annotation.KafkaListener(id = "${jc.consumer.group.id}",topicPartitions =//配置topic和分区{ @TopicPartition(topic = TopicConfig.TOPIC_JC_KAFKA_DEMO, partitions ="#{'${jc.consumer.partitions}'.split(',')}")})public void receive(@Payload List<String> messages,@Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions,@Header(KafkaHeaders.OFFSET) List<Long> offsets, Acknowledgment ack) {for (int i = 0; i < messages.size(); i++) {String msg = "message='" + messages.get(i) + "' with partition-offset='" + partitions.get(i) + "-" + offsets.get(i) + "'";logger.debug("receive messages {}",msg);}ack.acknowledge();logger.debug("all batch messages {} consumed",messages.size());}}

消费者监听类

package com.jc.demo.springboot.listener;import com.jc.demo.springboot.config.TopicConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;import java.util.List;public class KafkaListener {private static Logger logger = LoggerFactory.getLogger(KafkaListener.class);@org.springframework.kafka.annotation.KafkaListener(id = "${jc.consumer.group.id}",topicPartitions =//配置topic和分区{ @TopicPartition(topic = TopicConfig.TOPIC_JC_KAFKA_DEMO, partitions ="#{'${jc.consumer.partitions}'.split(',')}")})public void receive(@Payload List<String> messages,@Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions,@Header(KafkaHeaders.OFFSET) List<Long> offsets, Acknowledgment ack) {for (int i = 0; i < messages.size(); i++) {String msg = "message='" + messages.get(i) + "' with partition-offset='" + partitions.get(i) + "-" + offsets.get(i) + "'";logger.debug("receive messages {}",msg);}ack.acknowledge();logger.debug("all batch messages {} consumed",messages.size());}}

运行kafaka

07-15 19:39:25 [jc-consumer-group-1-0-C-1] DEBUG com.jc.demo.springboot.listener.KafkaListener - receive messages message='jevoncode1531654765409' with partition-offset='0-17'
07-15 19:39:25 [jc-consumer-group-1-0-C-1] DEBUG com.jc.demo.springboot.listener.KafkaListener - all batch messages 1 consumed

总结

kafka的使用、以上过程完成了基本的kafak的使用

小明菜市场

推荐阅读

● 搭建 | 一步成功搭建Centos + Kubernetes 环境

kafka入门教程。● 震惊 | 某公司实习生跑路，竟为了学习偷盗面试题

● 警钟 | 还不会Spring Boot集成JWT，你可能错过了大厂的Office了

● 图例 | Java混合模式分析之火焰图实例

● 面试 | 从一个API缓存演化，详细了解Redis各项功能

kafka hadoop？                                       

 

版权声明：本站所有资料均为网友推荐收集整理而来,仅供学习和研究交流使用。

原文链接：https://hbdhgg.com/1/74609.html

上一篇：这是小小本周的第六篇，本篇小小将会介绍一个很古老很古老很古老的为什么系列之不能重写service方法。...
下一篇：警钟 | 还不会Spring Boot集成JWT，你可能错过了大厂的Office了
标签：kafka快速入门kafka应用slf4j kafkakafka实战kafka与hadoop的区别kafka的使用kafka入门教程kafka hadoop
kafka快速入门

发表评论:

最新文章
阅读排行
猜你喜欢
热门标签

本站为非赢利网站，部分文章来源或改编自互联网及其他公众平台，主要目的在于分享信息，版权归原作者所有，内容仅供读者参考，如有侵权请联系我们删除！

Copyright © 2022 匯編語言學習筆記 Inc. 保留所有权利。

底部版权信息