kafka
1)kafka介绍
Kafka 是一个分布式流媒体平台,类似于消息队列或企业消息传递系统。kafka官网:http://kafka.apache.org/
2)kafka概述
消息中间件对比
特性ActiveMQRabbitMQRocketMQKafka开发语言javaerlangjavascala单机吞吐量万级万级10万级100万级时效性msusmsms级以内可用性高(主从)高(主从)非常高(分布式)非常高(分布式)功能特性成熟的产品、较全的文档、各种协议支持好并发能力强、性能好、延迟低MQ功能比较完善,扩展性佳只支持主要的MQ功能,主要应用于大数据领域
消息中间件对比-选择建议
消息中间件****建议Kafka追求高吞吐量,适合产生大量数据的互联网服务的数据收集业务RocketMQ可靠性要求很高的金融互联网领域,稳定性高,经历了多次阿里双11考验RabbitMQ性能较好,社区活跃度高,数据量没有那么大,优先选择功能比较完备的RabbitMQ
Kafka 基础架构
kafka介绍-名词解释
- producer:发布消息的对象称之为主题生产者(Kafka topic producer)
- partition:分区,每个主题可以创建多个分区,每个分区都由一系列有序和不可变的消息组成
- topic:Kafka将消息分门别类,每一类的消息称之为一个主题(Topic)
- consumer:订阅消息并处理发布的消息的对象称之为主题消费者(consumers)
- broker:已发布的消息保存在一组服务器中,称之为Kafka集群。集群中的每一个服务器都是一个代理(Broker)。 消费者可以订阅一个或多个主题(topic),并从Broker拉数据,从而消费这些已发布的消息。
- record:消息,消息队列基础通信单位
- replica:副本,每个分区都有一个至多个副本存在,它的主要作用是存储保存数据,以日志(Log)对象的形式体现。副本又分为leader副本和follower副本
- offset:偏移量,每一个消息在日志文件中的位置都对应一个按序递增的偏移量,你可以理解为类似数组的存储形式
3)kafka安装配置
Kafka对于zookeeper是强依赖,保存kafka相关的节点数据,所以安装Kafka之前必须先安装zookeeper
docker安装
- Docker安装zookeeper
下载镜像:
docker pull zookeeper:3.4.14
创建容器
docker run -d --name zookeeper -p 2181:2181 zookeeper:3.4.14
- Docker安装kafka
下载镜像:
docker pull wurstmeister/kafka:2.12-2.3.1
创建容器
docker run -d --name kafka \
--env KAFKA_ADVERTISED_HOST_NAME=192.168.200.130 \
--env KAFKA_ZOOKEEPER_CONNECT=192.168.200.130:2181 \
--env KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.200.130:9092 \
--env KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 \
--env KAFKA_HEAP_OPTS="-Xmx256M -Xms256M"\
--net=host wurstmeister/kafka:2.12-2.3.1
传统安装
Zookeeper安装教程
官网下载地址:https://link.csdn.net/?target=https%3A%2F%2Farchive.apache.org%2Fdist%2Fzookeeper%2F
安装步骤
1、打开官网下载对应版本,复制到服务器文件夹中
2、使用tar -zxvf 命令解压安装包 [root@VM-4-3-centos conf]# tar -zxvf apache-zookeeper-3.5.7-bin.tar.gz
3、使用cd 命令进入解压后的目录,进入conf文件夹,使用mv命令修改zoo-sample.cfg文件名为zoo.cfg,简化文件名称。
4、使用vim 命令进入zoo.cfg 修改Zookeeper数据存储地址,默认地址为Linux临时文件目录,在安装目录下新建zkData文件夹,地址切换为zkData地址,修改后保存并退出。
5、进入bin目录下使用 ./zkServer.sh start 启动服务 ,启动完成后使用 jps 命令查看Java进程是否启动成功,未安装的话使用yum install -y java-1.8.0-openjdk-devel.x86_64 命令进行安装
6、启动完服务端可使用命令 ./zkCli.sh 启动Zookeeper客户端进行访问,使用 quit 命令可退出Zookeeper客户端
7、Zookeeper conf参数详解 :
tickTime = 2000 :通信心跳时间,Zookeeper服务器与客户端心跳时间,单位毫秒
initLimit = 10 :LF初始通信时限
syncLimit = 5 :LF同步通信时限
dataDir:保存Zookeeper中的数据
clientPort = 2181:客户端连接端口,通常不做修改
zookeeper集群配置
** 2.1.1** 集群规划
masterslave0slave1zookeeperzookeeperzookeeperkafkakafkakafka123
1、在Zookeeper数据存储地址文件夹中使用 vim myid 新建一个myid文件(切记必须为myid文件名),然后在里面输入一个数字(切记该数字在集群中必须唯一不可重复)然后保存并退出。
2、如有集群服务器便将该安装文件夹分发到其它服务器下,然后修改该文件myid(数值不可重复,切记)
3、在配置文件中末尾增加如下配置(根据自身情况)
4、配置说明
5、配置完成后启动bin目录下的zkServer.sh 服务,切记,集群服务必须启动半数以上服务才能启动成功,可使用zkServier.sh status 查看当前Zookeeper状态
6.查看zookeeper启动状态,脚本
#!/bin/bashforiin master slave0 slave1;do#statementsecho"------------------$i----------------------"ssh$i"source /etc/profile;jps $@ | grep -v Jps"done
Kafka安装教程
官网下载地址:Apache Kafka
Kafka3.0.0版本不再支持Java8,Kafka2.8.0可替换不使用ZooKeeper:
安装步骤
1、打开官网下载所需要的版本,复制到服务器文件夹中
2、使用 tar -zxvf 解压下载的安装包 [root@VM-4-3-centos kafka]# tar -zxvf kafka_2.13-2.8.1.tgz
3、使用cd 命令进入Kafka config配置文件夹
4、使用vim 命令进入server.properties文件,修改配置参数broker.id=0 ,该数值在集群服务中为唯一,不允许重复
5、往下翻,找到配置参数log.dirs=/tmp/kafka-logs,改参数地址为存储Kafka数据地址,默认地址为Linux临时目录,会不定时回收,请修改地址
6、往下翻,找到配置参数zookeeper.connect=localhost:2181,该配置参数为zookeepe集群的地址,可以是多个,多个之间用逗号分割,一般端口都为2181;master:2181,slave0:2181,slave1:2181。完成后使用:wq命令保存退出
7、配置Kafka环境变量,使用cd命令进入根目录,使用vim ./etc/profile 文件配置Kafkal路径并保存退出 后面的路径为kafka的文件夹地址
export KAFKA_HOME=/opt/kafka_2.12-3.2.3
export PATH=
P
A
T
H
:
PATH:
PATH:KAFKA_HOME/bin
8、使用./kafka-server-start.sh -daemon …/config/server.properties 启动Kafka服务,切记,启动Kafka服务前必须启动Zookeeper服务。然后使用jps命令查看状态是否启动
kafka集群配置
(1)先启动 Zookeeper 集群,然后启动 Kafka。
启动脚本
#!/bin/bashcase$1in"start"){foriin master slave0 slave1
doecho ------------- zookeeper $i 启动 ------------
ssh$i"/software/apache-zookeeper-3.6.3-bin/bin/zkServer.sh start"done};;"stop"){foriin master slave0 slave1
doecho ------------- zookeeper $i 停止 ------------
ssh$i"/software/apache-zookeeper-3.6.3-bin/bin/zkServer.sh stop"done};;"status"){foriin master slave0 slave1
doecho ------------- zookeeper $i 状态 ------------
ssh$i"/software/apache-zookeeper-3.6.3-bin/bin/zkServer.sh status"done};;esac
(2)依次在 master、slave0、slave1节点上启动 Kafka。
使用./kafka-server-start.sh -daemon …/config/server.properties 启动Kafka服务
(3)查看启动状态
Zookeeper关闭前必须先关闭Kafka,否则将无法再关闭Kafka(当然可以强制杀死)
如果不是本地访问,需要修改kafka配置文件
问题:
在服务器本地可以使用命令行参数连接并且进行发布订阅操作,但是使用spring boot连接服务器则不行,会有如下错误:
Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available
这个问题需要我们修改下kafka的配置文件,添加一个对外暴露的ip地址,供我们连接使用。
修改server.properties的两行默认配置,即可通过外网连接服务器Kafka,问题解决:
# 允许外部端口连接
listeners=PLAINTEXT://0.0.0.0:9092
# 外部代理地址
advertised.listeners=PLAINTEXT://192.168.64.12:9092
4)kafka入门
- 生产者发送消息,多个消费者只能有一个消费者接收到消息
- 生产者发送消息,多个消费者都可以接收到消息
(1)创建kafka-demo项目,导入依赖
<dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency>
(2)配置文件
kafka:#kafka的服务地址bootstrap-servers: 192.168.20.83:9092producer:# producer 生产者retries:0# 重试次数acks:1# 应答级别:多少个分区副本备份完成时向生产者发送ack确认(可选0、1、all/-1)batch-size:16384# 批量大小buffer-memory:33554432# 生产端缓冲区大小key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer
consumer:# consumer消费者group-id: rcgroup # 默认的消费组IDenable-auto-commit:true# 是否自动提交offsetauto-commit-interval:100# 提交offset延时(接收到消息后多久提交offset)# earliest:当各分区下有已提交的offset时,从提交的offset开始消费;无提交的offset时,从头开始消费# latest:当各分区下有已提交的offset时,从提交的offset开始消费;无提交的offset时,消费新产生的该分区下的数据# none:topic各分区都存在已提交的offset时,从offset后开始消费;只要有一个分区不存在已提交的offset,则抛出异常auto-offset-reset: latest
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
(2)生产者发送消息
importorg.apache.kafka.clients.producer.KafkaProducer;importorg.apache.kafka.clients.producer.ProducerConfig;importorg.apache.kafka.clients.producer.ProducerRecord;importjava.util.Properties;@TestpublicvoidkafkaTest(){String topic ="Rc-topic";String msg ="发送了一条消息";ListenableFuture<SendResult<String,String>> send = kafkaTemplate.send(topic, msg);System.out.println(send);}}
(3)消费者接收消息
@ComponentpublicclassMessageListener{@KafkaListener(topics ={"Rc-topic"},groupId ="rcGroup")publicvoidlistener(ConsumerRecord<String,String>record){//获取消息String message =record.value();//消息偏移量long offset =record.offset();System.out.println("读取的消息:"+message+"\n当前偏移量:"+offset);}}
(4)发送成功
总结
- 生产者发送消息,多个消费者订阅同一个主题,只能有一个消费者收到消息(一对一)
- 生产者发送消息,多个消费者订阅同一个主题,所有消费者都能收到消息(一对多)
5)kafka高可用设计
5.1)集群
- Kafka 的服务器端由被称为 Broker 的服务进程构成,即一个 Kafka 集群由多个 Broker 组成
- 这样如果集群中某一台机器宕机,其他机器上的 Broker 也依然能够对外提供服务。这其实就是 Kafka 提供高可用的手段之一
5.2)备份机制(Replication)
Kafka 中消息的备份又叫做 副本(Replica)
Kafka 定义了两类副本:
- 领导者副本(Leader Replica)
- 追随者副本(Follower Replica)
同步方式
ISR(in-sync replica)需要同步复制保存的follower
如果leader失效后,需要选出新的leader,选举的原则如下:
第一:选举时优先从ISR中选定,因为这个列表中follower的数据是与leader同步的
第二:如果ISR列表中的follower都不行了,就只能从其他follower中选取
极端情况,就是所有副本都失效了,这时有两种方案
第一:等待ISR中的一个活过来,选为Leader,数据可靠,但活过来的时间不确定
第二:选择第一个活过来的Replication,不一定是ISR中的,选为leader,以最快速度恢复可用性,但数据不一定完整
6)kafka生产者详解
6.1)发送类型
- 同步发送使用send()方法发送,它会返回一个Future对象,调用get()方法进行等待,就可以知道消息是否发送成功
RecordMetadata recordMetadata = producer.send(kvProducerRecord).get();System.out.println(recordMetadata.offset());
- 异步发送调用send()方法,并指定一个回调函数,服务器在返回响应时调用函数
//异步消息发送
producer.send(kvProducerRecord,newCallback(){@OverridepublicvoidonCompletion(RecordMetadata recordMetadata,Exception e){if(e !=null){System.out.println("记录异常信息到日志表中");}System.out.println(recordMetadata.offset());}});
6.2)参数详解
- ack
代码的配置方式:
//ack配置 消息确认机制
prop.put(ProducerConfig.ACKS_CONFIG,"all");
参数的选择说明
ACK应答级别:
确认机制****说明acks=0生产者在成功写入消息之前不会等待任何来自服务器的响应,消息有丢失的风险,但是速度最快acks=1(默认值)只要集群首领节点收到消息,生产者就会收到一个来自服务器的成功响应acks=all只有当所有参与赋值的节点全部收到消息时,生产者才会收到一个来自服务器的成功响应
可靠性总结:
*acks=0***,生产者发送过来数据就不管了,可靠性差,效率高;
*acks=1***,生产者发送过来数据Leader应答,可靠性中等,效率中等;
*acks=-1***,生产者发送过来数据Leader和ISR队列里面所有Follwer应答,可靠性高,效率低;
*在生产环境中,acks=0很少使用;acks=1,一般用于传输普通日志,允许丢个别数据;acks=-1***,一般用于传输和钱相关的数据,
对可靠性要求比较高的场景。
- retries
生产者从服务器收到的错误有可能是临时性错误,在这种情况下,retries参数的值决定了生产者可以重发消息的次数,如果达到这个次数,生产者会放弃重试返回错误,默认情况下,生产者会在每次重试之间等待100ms
代码中配置方式:
//重试次数
prop.put(ProducerConfig.RETRIES_CONFIG,10);
- 消息压缩
默认情况下, 消息发送时不会被压缩。
代码中配置方式:
//数据压缩
prop.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"lz4");
压缩算法****说明snappy占用较少的 CPU, 却能提供较好的性能和相当可观的压缩比, 如果看重性能和网络带宽,建议采用lz4占用较少的 CPU, 压缩和解压缩速度较快,压缩比也很客观gzip占用较多的 CPU,但会提供更高的压缩比,网络带宽有限,可以使用这种算法
使用压缩可以降低网络传输开销和存储开销,而这往往是向 Kafka 发送消息的瓶颈所在。
7)kafka消费者详解
7.1)消费者组
- 消费者组(Consumer Group) :指的就是由一个或多个消费者组成的群体
- 一个发布在Topic上消息被分发给此消费者组中的一个消费者- 所有的消费者都在一个组中,那么这就变成了queue模型- 所有的消费者都在不同的组中,那么就完全变成了发布-订阅模型
7.2)消息有序性
应用场景:
- 即时消息中的单对单聊天和群聊,保证发送方消息发送顺序与接收方的顺序一致
- 充值转账两个渠道在同一个时间进行余额变更,短信通知必须要有顺序
topic分区中消息只能由消费者组中的唯一一个消费者处理,所以消息肯定是按照先后顺序进行处理的。但是它也仅仅是保证Topic的一个分区顺序处理,不能保证跨分区的消息先后处理顺序。 所以,如果你想要顺序的处理Topic的所有消息,那就只提供一个分区。
7.3)提交和偏移量
kafka不会像其他JMS队列那样需要得到消费者的确认,消费者可以使用kafka来追踪消息在分区的位置(偏移量)
消费者会往一个叫做_consumer_offset的特殊主题发送消息,消息里包含了每个分区的偏移量。如果消费者发生崩溃或有新的消费者加入群组,就会触发再均衡
正常的情况
如果消费者2挂掉以后,会发生再均衡,消费者2负责的分区会被其他消费者进行消费
再均衡后不可避免会出现一些问题
问题一:
如果提交偏移量小于客户端处理的最后一个消息的偏移量,那么处于两个偏移量之间的消息就会被重复处理。
问题二:
如果提交的偏移量大于客户端的最后一个消息的偏移量,那么处于两个偏移量之间的消息将会丢失。
如果想要解决这些问题,还要知道目前kafka提交偏移量的方式:
提交偏移量的方式有两种,分别是自动提交偏移量和手动提交
- 自动提交偏移量
当enable.auto.commit被设置为true,提交方式就是让消费者自动提交偏移量,每隔5秒消费者会自动把从poll()方法接收的最大偏移量提交上去
- 手动提交 ,当enable.auto.commit被设置为false可以有以下三种提交方式- 提交当前偏移量(同步提交)- 异步提交- 同步和异步组合提交
1.提交当前偏移量(同步提交)
把
enable.auto.commit
设置为false,让应用程序决定何时提交偏移量。使用commitSync()提交偏移量,commitSync()将会提交poll返回的最新的偏移量,所以在处理完所有记录后要确保调用了commitSync()方法。否则还是会有消息丢失的风险。
只要没有发生不可恢复的错误,commitSync()方法会一直尝试直至提交成功,如果提交失败也可以记录到错误日志里。
while(true){ConsumerRecords<String,String> records = consumer.poll(Duration.ofMillis(1000));for(ConsumerRecord<String,String>record: records){System.out.println(record.value());System.out.println(record.key());try{
consumer.commitSync();//同步提交当前最新的偏移量}catch(CommitFailedException e){System.out.println("记录提交失败的异常:"+e);}}}
2.异步提交
手动提交有一个缺点,那就是当发起提交调用时应用会阻塞。当然我们可以减少手动提交的频率,但这个会增加消息重复的概率(和自动提交一样)。另外一个解决办法是,使用异步提交的API。
while(true){ConsumerRecords<String,String> records = consumer.poll(Duration.ofMillis(1000));for(ConsumerRecord<String,String>record: records){System.out.println(record.value());System.out.println(record.key());}
consumer.commitAsync(newOffsetCommitCallback(){@OverridepublicvoidonComplete(Map<TopicPartition,OffsetAndMetadata> map,Exception e){if(e!=null){System.out.println("记录错误的提交偏移量:"+ map+",异常信息"+e);}}});}
3.同步和异步组合提交
异步提交也有个缺点,那就是如果服务器返回提交失败,异步提交不会进行重试。相比较起来,同步提交会进行重试直到成功或者最后抛出异常给应用。异步提交没有实现重试是因为,如果同时存在多个异步提交,进行重试可能会导致位移覆盖。
举个例子,假如我们发起了一个异步提交commitA,此时的提交位移为2000,随后又发起了一个异步提交commitB且位移为3000;commitA提交失败但commitB提交成功,此时commitA进行重试并成功的话,会将实际上将已经提交的位移从3000回滚到2000,导致消息重复消费。
try{while(true){ConsumerRecords<String,String> records = consumer.poll(Duration.ofMillis(1000));for(ConsumerRecord<String,String>record: records){System.out.println(record.value());System.out.println(record.key());}
consumer.commitAsync();}}catch(Exception e){+
e.printStackTrace();System.out.println("记录错误信息:"+e);}finally{try{
consumer.commitSync();}finally{
consumer.close();}}
8)幂等性
8.1幂等性原理
幂等性就是指Producer不论向Broker发送多少次重复数据,Broker端都只会持久化一条,保证了不重复。
**精确一次(Exactly Once)= 幂等性 + 至少一次( ack=-1 +**分区副本数>=2 + ISR最小副本数量>=2) 。
重复数据的判断标准:具有<PID, Partition, SeqNumber>相同主键的消息提交时,Broker只会持久化一条。其
中PID是Kafka每次重启都会分配一个新的;Partition 表示分区号;Sequence Number是单调自增的。
所以幂等性只能保证的是在单分区单会话内不重复
8.2)如何使用幂等性
开启参数 enable.idempotence 默认为 true,false 关闭。
9)springboot集成kafka
9.1)入门
1.导入spring-kafka依赖信息
<dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!-- kafkfa --><dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId><exclusions><exclusion><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId></exclusion></exclusions></dependency><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId></dependency></dependencies>
2.在resources下创建文件application.yml
server:port:9991spring:application:name: kafka-demo
kafka:bootstrap-servers: 192.168.200.130:9092producer:retries:10key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer
consumer:group-id: ${spring.application.name}-test
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
3.消息生产者
importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.kafka.core.KafkaTemplate;importorg.springframework.web.bind.annotation.GetMapping;importorg.springframework.web.bind.annotation.RestController;@RestControllerpublicclassHelloController{@AutowiredprivateKafkaTemplate<String,String> kafkaTemplate;@GetMapping("/hello")publicStringhello(){
kafkaTemplate.send("rc-topic","任晨测试");return"ok";}}
4.消息消费者
importorg.springframework.kafka.annotation.KafkaListener;importorg.springframework.stereotype.Component;importorg.springframework.util.StringUtils;@ComponentpublicclassHelloListener{@KafkaListener(topics ="rc-topic")publicvoidonMessage(String message){if(!StringUtils.isEmpty(message)){System.out.println(message);}}}
9.2)传递消息为对象
目前springboot整合后的kafka,因为序列化器是StringSerializer,这个时候如果需要传递对象可以有两种方式
方式一:可以自定义序列化器,对象类型众多,这种方式通用性不强,本章节不介绍
方式二:可以把要传递的对象进行转json字符串,接收消息后再转为对象即可,本项目采用这种方式
- 发送消息
@GetMapping("/hello")publicStringhello(){User user =newUser();
user.setUsername("xiaowang");
user.setAge(18);
kafkaTemplate.send("user-topic", JSON.toJSONString(user));return"ok";}
- 接收消息
importcom.alibaba.fastjson.JSON;importcom.rc.kafka.pojo.User;importorg.springframework.kafka.annotation.KafkaListener;importorg.springframework.stereotype.Component;importorg.springframework.util.StringUtils;@ComponentpublicclassHelloListener{@KafkaListener(topics ="user-topic")publicvoidonMessage(String message){if(!StringUtils.isEmpty(message)){User user = JSON.parseObject(message,User.class);System.out.println(user);}}}
2 Kafka Stream
2.1 概述
Kafka Stream是Apache Kafka从0.10版本引入的一个新Feature。它是提供了对存储于Kafka内的数据进行流式处理和分析的功能。
Kafka Stream的特点如下:
- Kafka Stream提供了一个非常简单而轻量的Library,它可以非常方便地嵌入任意Java应用中,也可以任意方式打包和部署
- 除了Kafka外,无任何外部依赖
- 充分利用Kafka分区机制实现水平扩展和顺序性保证
- 通过可容错的state store实现高效的状态操作(如windowed join和aggregation)
- 支持正好一次处理语义
- 提供记录级的处理能力,从而实现毫秒级的低延迟
- 支持基于事件时间的窗口操作,并且可处理晚到的数据(late arrival of records)
- 同时提供底层的处理原语Processor(类似于Storm的spout和bolt),以及高层抽象的DSL(类似于Spark的map/group/reduce)
2.2 Kafka Streams的关键概念
- 源处理器(Source Processor):源处理器是一个没有任何上游处理器的特殊类型的流处理器。它从一个或多个kafka主题生成输入流。通过消费这些主题的消息并将它们转发到下游处理器。
- Sink处理器:sink处理器是一个没有下游流处理器的特殊类型的流处理器。它接收上游流处理器的消息发送到一个指定的Kafka主题。
2.3 KStream
(1)数据结构类似于map,如下图,key-value键值对
(2)KStream
KStream数据流(data stream),即是一段顺序的,可以无限长,不断更新的数据集。
数据流中比较常记录的是事件,这些事件可以是一次鼠标点击(click),一次交易,或是传感器记录的位置数据。
KStream负责抽象的,就是数据流。与Kafka自身topic中的数据一样,类似日志,每一次操作都是向其中插入(insert)新数据。
为了说明这一点,让我们想象一下以下两个数据记录正在发送到流中:
(“ alice”,1)->(“” alice“,3)
如果您的流处理应用是要总结每个用户的价值,它将返回
4
了
alice
。为什么?因为第二条数据记录将不被视为先前记录的更新。(insert)新数据
2.4 SpringBoot集成Kafka Stream 入门案例编写-求单词个数
(1)需求分析,求单词个数(word count)
消息的内容:hello kafka hello rc
结果:hello(2)
kafka(1)
rc(1)
(2)引入依赖
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-streams</artifactId>
</dependency>
(3)自定配置参数
importlombok.Getter;importlombok.Setter;importorg.apache.kafka.clients.consumer.ConsumerConfig;importorg.apache.kafka.common.serialization.Serdes;importorg.apache.kafka.streams.StreamsConfig;importorg.apache.kafka.streams.Topology;importorg.springframework.boot.context.properties.ConfigurationProperties;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;importorg.springframework.kafka.annotation.EnableKafkaStreams;importorg.springframework.kafka.annotation.KafkaStreamsDefaultConfiguration;importorg.springframework.kafka.config.KafkaStreamsConfiguration;importjava.util.HashMap;importjava.util.Map;/**
* 通过重新注册KafkaStreamsConfiguration对象,设置自定配置参数
*/@Setter@Getter@Configuration@EnableKafkaStreams@ConfigurationProperties(prefix="kafka")publicclassKafkaStreamConfig{privatestaticfinalint MAX_MESSAGE_SIZE =16*1024*1024;privateString hosts;privateString group;@Bean(name =KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)publicKafkaStreamsConfigurationdefaultKafkaStreamsConfig(){Map<String,Object> props =newHashMap<>();
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, hosts);
props.put(StreamsConfig.APPLICATION_ID_CONFIG,this.getGroup()+"_stream_aid");
props.put(StreamsConfig.CLIENT_ID_CONFIG,this.getGroup()+"_stream_cid");
props.put(StreamsConfig.RETRIES_CONFIG,10);
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG,Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,Serdes.String().getClass());returnnewKafkaStreamsConfiguration(props);}}
修改application.yml文件,在最下方添加自定义配置
kafka:hosts: 192.168.200.130:9092group: ${spring.application.name}
(4)新增配置类,创建KStream对象,进行聚合5
importlombok.extern.slf4j.Slf4j;importorg.apache.kafka.streams.KeyValue;importorg.apache.kafka.streams.StreamsBuilder;importorg.apache.kafka.streams.kstream.KStream;importorg.apache.kafka.streams.kstream.TimeWindows;importorg.apache.kafka.streams.kstream.ValueMapper;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;importjava.time.Duration;importjava.util.Arrays;@Configuration@Slf4jpublicclassKafkaStreamHelloListener{@BeanpublicKStream<String,String>kStream(StreamsBuilder streamsBuilder){//创建kstream对象,同时指定从那个topic中接收消息KStream<String,String> stream = streamsBuilder.stream("rc-topic-input");
stream.flatMapValues(newValueMapper<String,Iterable<String>>(){@OverridepublicIterable<String>apply(String value){returnArrays.asList(value.split(" "));}})//根据value进行聚合分组.groupBy((key,value)->value)//聚合计算时间间隔.windowedBy(TimeWindows.of(Duration.ofSeconds(10)))//求单词的个数.count().toStream()//处理后的结果转换为string字符串.map((key,value)->{System.out.println("key:"+key+",value:"+value);returnnewKeyValue<>(key.key().toString(),value.toString());})//发送消息.to("rc-topic-out");return stream;}}
(5) 发送消息到 “rc-topic-input”
@TestpublicvoidkafkaTest()throwsExecutionException,InterruptedException{String topic ="rc-topic-input";String msg ="hello kafka hello rc";SendResult<String,String> stringStringSendResult = kafkaTemplate.send(topic, msg).get();System.out.println(stringStringSendResult);}
(6)接收消息
packagecn.amarone.listen;importorg.apache.kafka.clients.consumer.ConsumerRecord;importorg.springframework.kafka.annotation.KafkaListener;importorg.springframework.stereotype.Component;@ComponentpublicclassMessageListener{@KafkaListener(topics ={"itcast-topic-out"},groupId ="rcGroup")publicvoidlistener(ConsumerRecord<String,String>record){//获取消息String message =record.value();//消息偏移量long offset =record.offset();System.out.println("读取的消息:"+message+"\n当前偏移量:"+offset);}}
测试:
.to(“rc-topic-out”);
return stream;
}
}
(5) 发送消息到 “rc-topic-input”
```java
@Test
public void kafkaTest() throws ExecutionException, InterruptedException {
String topic = "rc-topic-input";
String msg = "hello kafka hello rc";
SendResult<String, String> stringStringSendResult = kafkaTemplate.send(topic, msg).get();
System.out.println(stringStringSendResult);
}
(6)接收消息
packagecn.amarone.listen;importorg.apache.kafka.clients.consumer.ConsumerRecord;importorg.springframework.kafka.annotation.KafkaListener;importorg.springframework.stereotype.Component;@ComponentpublicclassMessageListener{@KafkaListener(topics ={"itcast-topic-out"},groupId ="rcGroup")publicvoidlistener(ConsumerRecord<String,String>record){//获取消息String message =record.value();//消息偏移量long offset =record.offset();System.out.println("读取的消息:"+message+"\n当前偏移量:"+offset);}}
[外链图片转存中…(img-ca7Eah8P-1664521091420)]
测试:
启动微服务,正常发送消息,可以正常接收到消息
版权归原作者 你的猪脑我另做他用 所有, 如有侵权,请联系我们删除。