kafka安装及入门
kafka
1)kafka介绍
Kafka 是一个分布式流媒体平台,类似于消息队列或企业消息传递系统。kafka官网:http://kafka.apache.org/
2)kafka概述
消息中间件对比
特性 | ActiveMQ | RabbitMQ | RocketMQ | Kafka |
---|---|---|---|---|
开发语言 | java | erlang | java | scala |
单机吞吐量 | 万级 | 万级 | 10万级 | 100万级 |
时效性 | ms | us | ms | ms级以内 |
可用性 | 高(主从) | 高(主从) | 非常高(分布式) | 非常高(分布式) |
功能特性 | 成熟的产品、较全的文档、各种协议支持好 | 并发能力强、性能好、延迟低 | 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.gz3、使用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** 集群规划
master | slave0 | slave1 |
---|---|---|
zookeeper | zookeeper | zookeeper |
kafka | kafka | kafka |
1 | 2 | 3 |
1、在Zookeeper数据存储地址文件夹中使用 vim myid 新建一个myid文件(切记必须为myid文件名),然后在里面输入一个数字(切记该数字在集群中必须唯一不可重复)然后保存并退出。
2、如有集群服务器便将该安装文件夹分发到其它服务器下,然后修改该文件myid(数值不可重复,切记)
3、在配置文件中末尾增加如下配置(根据自身情况)
4、配置说明
5、配置完成后启动bin目录下的zkServer.sh 服务,切记,集群服务必须启动半数以上服务才能启动成功,可使用zkServier.sh status 查看当前Zookeeper状态
6.查看zookeeper启动状态,脚本
#!/bin/bash
for i in 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=PATH:PATH:PATH:KAFKA_HOME/bin
8、使用./kafka-server-start.sh -daemon …/config/server.properties 启动Kafka服务,切记,启动Kafka服务前必须启动Zookeeper服务。然后使用jps命令查看状态是否启动
kafka集群配置
(1)先启动 Zookeeper 集群,然后启动 Kafka。
启动脚本
#!/bin/bash
case $1 in
"start"){for i in master slave0 slave1doecho ------------- zookeeper $i 启动 ------------ssh $i "/software/apache-zookeeper-3.6.3-bin/bin/zkServer.sh start"done
}
;;
"stop"){for i in master slave0 slave1doecho ------------- zookeeper $i 停止 ------------ssh $i "/software/apache-zookeeper-3.6.3-bin/bin/zkServer.sh stop"done
}
;;
"status"){for i in master slave0 slave1doecho ------------- 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.StringSerializervalue-serializer: org.apache.kafka.common.serialization.StringSerializerconsumer: # 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: latestkey-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.apache.kafka.common.serialization.StringDeserializer
(2)生产者发送消息
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;import java.util.Properties;@Testpublic void kafkaTest(){String topic = "Rc-topic";String msg = "发送了一条消息";ListenableFuture<SendResult<String, String>> send = kafkaTemplate.send(topic, msg);System.out.println(send);}}
(3)消费者接收消息
@Component
public class MessageListener {@KafkaListener(topics = {"Rc-topic"},groupId = "rcGroup")public void listener(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, new Callback() {@Overridepublic void onCompletion(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(new OffsetCommitCallback() {@Overridepublic void onComplete(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: 9991
spring:application:name: kafka-demokafka:bootstrap-servers: 192.168.200.130:9092producer:retries: 10key-serializer: org.apache.kafka.common.serialization.StringSerializervalue-serializer: org.apache.kafka.common.serialization.StringSerializerconsumer:group-id: ${spring.application.name}-testkey-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.apache.kafka.common.serialization.StringDeserializer
3.消息生产者
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;@RestController
public class HelloController {@Autowiredprivate KafkaTemplate<String,String> kafkaTemplate;@GetMapping("/hello")public String hello(){kafkaTemplate.send("rc-topic","任晨测试");return "ok";}
}
4.消息消费者
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;@Component
public class HelloListener {@KafkaListener(topics = "rc-topic")public void onMessage(String message){if(!StringUtils.isEmpty(message)){System.out.println(message);}}
}
9.2)传递消息为对象
目前springboot整合后的kafka,因为序列化器是StringSerializer,这个时候如果需要传递对象可以有两种方式
方式一:可以自定义序列化器,对象类型众多,这种方式通用性不强,本章节不介绍
方式二:可以把要传递的对象进行转json字符串,接收消息后再转为对象即可,本项目采用这种方式
- 发送消息
@GetMapping("/hello")
public String hello(){User user = new User();user.setUsername("xiaowang");user.setAge(18);kafkaTemplate.send("user-topic", JSON.toJSONString(user));return "ok";
}
- 接收消息
import com.alibaba.fastjson.JSON;
import com.rc.kafka.pojo.User;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;@Component
public class HelloListener {@KafkaListener(topics = "user-topic")public void onMessage(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)自定配置参数
import lombok.Getter;
import lombok.Setter;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.Topology;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafkaStreams;
import org.springframework.kafka.annotation.KafkaStreamsDefaultConfiguration;
import org.springframework.kafka.config.KafkaStreamsConfiguration;import java.util.HashMap;
import java.util.Map;/*** 通过重新注册KafkaStreamsConfiguration对象,设置自定配置参数*/@Setter
@Getter
@Configuration
@EnableKafkaStreams
@ConfigurationProperties(prefix="kafka")
public class KafkaStreamConfig {private static final int MAX_MESSAGE_SIZE = 16* 1024 * 1024;private String hosts;private String group;@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)public KafkaStreamsConfiguration defaultKafkaStreamsConfig() {Map<String, Object> props = new HashMap<>();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());return new KafkaStreamsConfiguration(props);}
}
修改application.yml文件,在最下方添加自定义配置
kafka:hosts: 192.168.200.130:9092group: ${spring.application.name}
(4)新增配置类,创建KStream对象,进行聚合5
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.apache.kafka.streams.kstream.ValueMapper;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.time.Duration;
import java.util.Arrays;@Configuration
@Slf4j
public class KafkaStreamHelloListener {@Beanpublic KStream<String,String> kStream(StreamsBuilder streamsBuilder){//创建kstream对象,同时指定从那个topic中接收消息KStream<String, String> stream = streamsBuilder.stream("rc-topic-input");stream.flatMapValues(new ValueMapper<String, Iterable<String>>() {@Overridepublic Iterable<String> apply(String value) {return Arrays.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);return new KeyValue<>(key.key().toString(),value.toString());})//发送消息.to("rc-topic-out");return stream;}
}
(5) 发送消息到 “rc-topic-input”
@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)接收消息
package cn.amarone.listen;import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;@Component
public class MessageListener {@KafkaListener(topics = {"itcast-topic-out"},groupId = "rcGroup")public void listener(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)接收消息
package cn.amarone.listen;import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;@Component
public class MessageListener {@KafkaListener(topics = {"itcast-topic-out"},groupId = "rcGroup")public void listener(ConsumerRecord<String,String> record){//获取消息String message = record.value();//消息偏移量long offset = record.offset();System.out.println("读取的消息:"+message+"\n当前偏移量:"+offset);}}
[外链图片转存中…(img-ca7Eah8P-1664521091420)]
测试:
启动微服务,正常发送消息,可以正常接收到消息
kafka安装及入门相关推荐
- Kafka安装启动入门教程
前些天发现了一个巨牛的人工智能学习网站,通俗易懂,风趣幽默,忍不住分享一下给大家.点击跳转到网站:https://www.captainai.net/dongkelun 前言 本文讲如何安装启动kaf ...
- 《Kafka系列》Kafka详细教程入门
Kafka 1 消息队列--消息中间件 1.1 消息队列的作用 1.2 消息队列的概念--MQ Message 在互联网中,多台设备产生通信的数据的总称:可以是视频.文本.音频等等. Quene 一种 ...
- Kafka安装以及验证
文章目录 1 Kafka安装 1.1 下载安装 1.2 配置启动zookeeper 1.3 配置kafka 1.3.1 修改配置文件 1.3.2 配置环境变量 1.3.3 配置服务启动脚本 1.3.4 ...
- Kafka详解(上)——消息系统分类、Kafka安装、两种启动、基本概念、两种架构、核心配置文件
1 消息和消息系统 消息(Message)是指在应用间传送的数据.消息可以非常简单,比如只包含文本字符串,也可以更复杂,可能包含嵌入对象. 1-1 消息系统是什么 消息系统负责将数据从一个应用 ...
- Kafka消息队列 入门到精通 看这一篇就够了
文章目录 第一章 概述 1.1 Kafka 的定义及特点 1.2 消息队列的介绍 1.3 Kafka 的基础架构 第二章 入门 2.1 Kafka 的安装部署 2.2 Kafka 命令行操作 第三章 ...
- Python:Scrapy的安装和入门案例
Scrapy的安装介绍 Scrapy框架官方网址:http://doc.scrapy.org/en/latest Scrapy中文维护站点:http://scrapy-chs.readthedocs. ...
- windows下nodejs express安装及入门网站,视频资料,开源项目介绍
windows下nodejs express安装及入门网站,视频资料,开源项目介绍,pm2,supervisor,npm,Pomelo,Grunt安装使用注意事项等总结 第一步:下载安装文件 下载地址 ...
- TensorFlow 安装及入门
安装 入门 TensorFlow简介 基本操作 loading lazy 安装 上TensorFlow官网按照教程安装. 入门 以下学习内容follow这篇博客 TensorFlow简介 Tensor ...
- kudu接受kafka消息_Kafka入门详解
1.1 什么是kafka? Kafka最初由Linkedin公司开发,是一个分布式.支持分区的(partition).多副本的(replica),基于zookeeper协调的分布式消息系统,它的最大的 ...
最新文章
- linux校园网客户端,Ubuntu Linux环境下校园网客户端安装使用
- (GIT)代码分支管理策略
- 同步锁ReentrantLock
- php双层foreach,php 双层foreach 如何提升效率
- 永久开启完整版Google Play
- form表单提交,后台实体类接收转义问题
- eureka注册中心HA集群搭建
- 信息学奥赛C++语言:咪咪的智慧
- 第一次学游泳技巧_游泳前,让自己不再怕水的5堂准备课
- javascript window.history
- building a new horizon
- sql server 外键_什么是SQL Server中的外键
- 如何同时配置GitHub和gitlab的SSH
- 图文配置Silverlight3.0开发环境(转载)
- (亲测)使用cmd结束进程的3种方法
- eclipse的代码突然变成红色块或绿色块
- Mac常见问题:如何使用文件保险箱加密 Mac 上的启动磁盘!
- 利用Python开发一个微信定时发送器
- 深入Android应用开发_核心技术解析与最佳实践
- 项目报找不到属性时解决方法
热门文章
- 转载:Java 关于中文乱码问题的解决方案与经验
- 用Java实现简单的“马赛克拼图”
- xmind设置成中文_XMind中文版使用教程:创建和使用思维导图模板
- O奖得主也收藏的6大美赛数据库网站
- PHP cURL 返回 HTTP/1.1 100 Continue
- 计算机一级操作题题库在线,全国计算机一级考试题库操作题
- 这样回答离职原因,一定没问题!
- 「创新场景50」年度最佳场景实践榜单重磅发布
- 目标检测(Object Detection)--Faster R-CNN算法
- JAVA 中文汉字转换阿拉伯数字(最终版,已修复了亿及亿以下的汉数字的转换,超过后的数字,暂时不支持)