springboot整合kafka入门

这篇具有很好参考价值的文章主要介绍了springboot整合kafka入门。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

kafka基本概念

springboot整合kafka入门

producer: 生产者,负责发布消息到kafka cluster(kafka集群)中。生产者可以是web前端产生的page view,或者是服务器日志,系统CPU、memory等。

consumer: 消费者,每个consumer属于一个特定的consuer group(可为每个consumer指定group name,若不指定group name则属于默认的group)。创建消费者时,要指定消费者接受的消息的topic,该消费者只会接受该topic的消息。

topic: 每条发布到Kafka集群的消息都有一个类别,这个类别被称为topic。(物理上不同topic的消息分开存储,逻辑上一个topic的消息虽然保存于一个或多个broker上但用户只需指定消息的topic即可生产或消费数据而不必关心数据存于何处)。

broker: kafka集群包含一个或多个服务器,这些服务器就叫做broker。

本机安装kafka测试

安装kafka(mac下)

kafka下载: 从官网下载 kafka_2.13-2.7.0.tgz,直接解压即可。

本机测试kafka

1、进入到kafka的解压目录,输入命令启动zookeeper:

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

复制

打开另一个终端输入命令启动kafka:

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

复制

2、服务启起来后,可以创建生产者和消费者了。 再打开另一个终端输入命令创建生产者:

./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test_topic

复制

broker-list: 参数指定生产者所使用的broker localhost: 9092 参数表示broker,这个broker为本机(127.0.0.1),且使用的端口是kafka的默认端口号是9092 topic: 参数表示生产者生产的消息的topic 为 “test_topic”

最后再打开另一个终端创建消费者:

./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test_topic --from-beginning

复制

bootstrap-server: 是指定consumer从哪里(broker)取出消息 topic: 指定消费者consumer取出的 topic 为“test_topic”的消息。 from-beginning: Kafka实际环境有可能会出现Consumer全部宕机,虽然基于Kafka的高可用特性,消费者群组中的消费者可以实现再均衡,所有Consumer不处理数据的情况很少,但是还是有可能会出现,此时就要求Consumer重启的时候能够读取在宕机期间Producer发送的数据。基于消费者订阅模式默认是无法实现的,因为只能订阅最新发送的数据。通过消费者命令行可以实现,只要在命令行中加上–from-beginning即可

3、都创建完了可以通过生产者输入消息,消费者来接收并显示消息,效果图如下:

springboot整合kafka入门

springboot整合kafka(IDEA)

注意: kafka要是部署在服务器的话,本机就 要和服务器之间能ping通。

1、创建springboot项目:

springboot整合kafka入门

springboot整合kafka入门

2、创建两个类,分别为生产者和消费者 项目目录结构:

springboot整合kafka入门

配置文件application.yml:(一般项目自动生成的是applicaiton.properties,但为了书写简便,改成yml)

spring:
  kafka:
    bootstrap-servers: 127.0.0.1:9092 #服务器的ip及端口,可以写多个,服务器之间用“:”间隔
    producer: #生产者配置
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer: #消费者配置
      group-id: test #设置消费者的组id
      enable-auto-commit: true
# auto-commit-interval: 1000
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

复制

springboot启动类入口,KafkaStudyApplication.java:

package com.study.kafka.kafka_study;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class KafkaStudyApplication { 
   

    public static void main(String[] args) { 
   
        SpringApplication.run(KafkaStudyApplication.class, args);
    }

}

复制

TestKafkaProducerController.java:(生产者)

package com.study.kafka.kafka_study;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController     //定义这是一个控制器,可以通过浏览器访问
@RequestMapping("/kafka")
public class TestKafkaProducerController { 

@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
//当在浏览器上输入http://localhost:8080/kafka/send?msg=abc,就会发送abc到服务器上去让消费者接收,msg对应下面的String msg
@RequestMapping("/producerSend")
public String send(String msg){ 

kafkaTemplate.send("test_topic", msg); //使用kafka模板发送信息
return "success";
}
}

复制

TestConsumer.java:(消费者)

package com.study.kafka.kafka_study;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
@Component
public class TestConsumer { 

/** * 定义此消费者接收topic为“test_topic”的消息,监听服务器上的kafka是否有相关的消息发过来 * @param record record变量代表消息本身,可以通过ConsumerRecord<?,?>类型的record变量来打印接收的消息的各种信息 * */
@KafkaListener(topics = "test_topic")
public void listen (ConsumerRecord<?, ?> record) throws Exception { 

System.out.printf("topic = %s, offset = %d, value = %s \n", record.topic(), record.offset(), record.value());
}
}

复制

测试

1、运行KafkaStudyApplication.java之后,终端上输入消息时,不仅终端上(服务器)运行的测试消费者能收到,IDEA上的程序也能收到。

springboot整合kafka入门

springboot整合kafka入门

2、在浏览器上输入http://localhost:8080/kafka/producerSend?msg=web world31231,不仅IDEA上的消费者能收到,在终端(服务器)上运行的测试消费者也能收到:(其中8080是tomcat服务器的端口,springboot默认下带的是tomcat)

springboot整合kafka入门文章来源地址https://www.toymoban.com/news/detail-469807.html

到了这里,关于springboot整合kafka入门的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 【Kafka】基本概念

    这个流派通常有⼀台服务器作为 Broker,所有的消息都通过它中转。⽣产者把消息发送给它就结束⾃⼰的任务了,Broker 则把消息主动推送给消费者(或者消费者主动轮询) 1.1.1 重topic 生产者生产的消息有topic,消费者订阅topic,在重topic的消息队列⾥必然需要topic的存在 1.1.2 轻

    2024年02月05日
    浏览(71)
  • Kafka基本概念

    Kafka是最初由Linkedin公司开发,是一个分布式、分区的、多副本的、多生产者、多订阅者,基于 zookeeper协调的分布式日志系统(也可以当做MQ系统),常见可以用于web/nginx日志、访问日志,消息服务等等。 主要应用场景是:日志收集系统和消息系统。 Kafka主要设计目标如下:

    2024年02月13日
    浏览(61)
  • Kafka - 深入了解Kafka基础架构:Kafka的基本概念

    我们首先了解一些Kafka的基本概念。 1)Producer :消息生产者,就是向kafka broker发消息的客户端 2)Consumer :消息消费者,向kafka broker获取消息的客户端 3)Consumer Group (CG):消费者组,由多个consumer组成。消费者组内每个消费者负责消费不同分区的数据,一个broker可以有多个

    2024年02月08日
    浏览(37)
  • kafka--技术文档-基本概念-《快速了解kafka》

    学习一种新的消息中间键,卡夫卡!!! 官网网址 Apache Kafka         Kafka是一种开源的分布式流处理平台,由Apache软件基金会开发,用Scala和Java编写。它是一个高吞吐量的分布式发布订阅消息系统,可以处理消费者在网站中的所有动作流数据。这种动作可以是网页浏览、

    2024年02月11日
    浏览(48)
  • Kafka的基本概念和架构

    1.1 定义 Kafka是一个开源的分布式事件流平台 ( Event Streaming Platform ),被广泛用于高性能数据管道、流分 析、数据集成和关键任务应用。 1.2 各消息队列的比较 目前比 较常见的消息队列产品主要有 Kafka 、 RabbitMQ 、 RocketMQ 等。 在大数据场景主要采用 Kafka 作为消息队列。在

    2024年02月01日
    浏览(45)
  • kafka--kafka的基本概念-topic和partition

    topic是逻辑概念 以Topic机制来对消息进行分类的,同一类消息属于同一个Topic,你可以将每个topic看成是一个消息队列。 生产者(producer)将消息发送到相应的Topic,而消费者(consumer)通过从Topic拉取消息来消费 kafka中是要求消费者主动拉取消息消费的,它并不会主动推送消息

    2024年02月12日
    浏览(45)
  • kafka学习-基本概念与简单实战

    目录 1、核心概念 消息和批次 Topic和Partition Replicas Offset broker和集群 生产者和消费者 2、开发实战 2.1、消息发送 介绍 代码实现 2.2、消息消费 介绍 代码实现 2.3、SpringBoot Kafka pom application.yaml KafkaConfig producer consumer         kafka的基本数据单元,由字节数组组成。可以理解

    2024年02月09日
    浏览(38)
  • springboot:整合Kafka

    依赖 yaml配置 简单demo 下面示例创建了一个生产者,发送消息到topic1,消费者监听topic1消费消息。监听器用@KafkaListener注解,topics表示监听的topic,支持同时监听多个,用英文逗号分隔。 KafkaTemplate调用send时 默认采用异步发送 ,如果需要同步获取发送结果,调用get方法 带回调

    2024年02月08日
    浏览(33)
  • springboot整合kafka-笔记

    这里我的springboot版本是2.3.8.RELEASE,使用的kafka-mq的版本是2.12 测试发送kafka消息-控制台日志

    2024年02月12日
    浏览(42)
  • Kafka 整合 SpringBoot

    1. 添加依赖 2. 配置文件 application.properties 3. 代码实现发送消息 1. 添加依赖 2. 配置文件 applicatio.properties  3. 代码实现消费消息 首先在 kafka 节点上创建 topic: 打开kafka节点服务器的终端,输入以下命令: 一些常用命令: 编写测试代码:ApplicationTests.java

    2024年02月07日
    浏览(42)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包