springboot整合kafka入门

springboot整合kafka入门springboot整合kafka入门kafka基本概念本机安装kafka测试安装kafka(mac下)本机测试kafkaspringboot整合kafka(IDEA)测试kafka基本概念producer:生产者,负责发布消息到kafkacluster(kafka集群)中。生产者可以是web前端产生的pageview,或者是服务器日志,系统CPU、memory等。consumer:消费者,每个consumer属于一个特定的consuergroup(可为每个consumer指定group

大家好,又见面了,我是你们的朋友全栈君。如果您正在找激活码,请点击查看最新教程,关注关注公众号 “全栈程序员社区” 获取激活教程,可能之前旧版本教程已经失效.最新Idea2022.1教程亲测有效,一键激活。

Jetbrains全家桶1年46,售后保障稳定

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

Jetbrains全家桶1年46,售后保障稳定

打开另一个终端输入命令启动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(IDEA)

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

1、创建springboot项目:
在这里插入图片描述在这里插入图片描述2、创建两个类,分别为生产者和消费者
项目目录结构:
在这里插入图片描述
配置文件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上的程序也能收到。
在这里插入图片描述在这里插入图片描述2、在浏览器上输入http://localhost:8080/kafka/producerSend?msg=web world31231,不仅IDEA上的消费者能收到,在终端(服务器)上运行的测试消费者也能收到:(其中8080是tomcat服务器的端口,springboot默认下带的是tomcat)
在这里插入图片描述

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。

发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/232030.html原文链接:https://javaforall.cn

【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛

【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...

(0)
blank

相关推荐

  • Java工程师要求_工程师到高级工程师要几年

    Java工程师要求_工程师到高级工程师要几年从事java后端开发4年及以上,具有根据业务需求进行独立设计的能力 熟练使用Spring及SpringMVC开发 熟练使用常用持久层框架Mybatis\Hibernate 熟悉java多线程及并发线程池使用 熟练使用Mysql基本操作,有一定sql优化经验 有使用Memcached/redis框架其中之一进行缓存开发的经验 会简单使用js/jquery等常用功能进行前端页面开发  、本科及以上学历…

  • 实战模拟│揭秘为啥年会你抽不到特等奖[通俗易懂]

    实战模拟│揭秘为啥年会你抽不到特等奖[通俗易懂]抽奖不只是概率性事件,有可能是确定性事件

  • H5+个推实现消息推送服务

    H5+个推实现消息推送服务网上看了几篇教程都是比较老的版本了,根据前人的智慧,然后自己摸索了下,简单几步实现了在手机上推送自定义的消息。首先,在个注册个账号,开发阶段使用个人注册即可,个推注册地址注册完进入配置页面,对应用进行配置,框中的几个地方要特别注意注册时会要求填一个包名,这个就是H5中对应的包名,一定要对应起来,否则收不到推送消息。或者查看应用的标识一定要和下面的标识对应起来,这样才能接收到信息。…

  • IntelliJ IDEA 可以使用中文了「建议收藏」

    IntelliJ IDEA 可以使用中文了「建议收藏」1.前言今天IntelliJIdea2020.1正式发布了!最大的一个亮点莫过于开始支持中文了。相信很多英语不好的同学已经期盼已久了。但是感觉登录界面感觉变丑了!建议把文章看完再去升级。2.汉化体验先来看看我的效果,首先要告诉你这是官方汉化包,并不是第三方!我大致看了一圈,个别还没有汉化完毕,可能没有找到合适的描述词汇吧,但是绝大部分已经完成了。这下很多同学可以更方便使…

  • C语言中的结构体,结构体中数组初始化与赋值

    C语言中的结构体,结构体中数组初始化与赋值原作地址:https://www.cnblogs.com/zhangfeionline/p/5913905.html最近写c语言中的结构体遇到了些问题,从网上找了些资料如下:结构体是连续存储的,但由于结构体中成员类型各异,所以会存在内存对齐问题,也就是内存里面会有空档,具体的对齐方式这里暂不讨论;1.结构体的定义和赋值结构体是可以直接初始化的,在定义的时候,就可

  • Spring batch教程 之 spring batch简介

    Spring batch教程 之 spring batch简介SpringBatch介绍在企业领域,有很多应用和系统需要在生产环境中使用批处理来执行大量的业务操作.批处理业务需要自动地对海量数据信息进行各种复杂的业务逻辑处理,同时具备极高的效率,不需要人工干预.执行这种操作通常根据时间事件(如月末统计,通知或信件),或者定期处理那些业务规则超级复杂,数据量非常庞大的业务,(如保险赔款确定,利率调整),也可能是从内部/外部系统抓取到的各种数据,通常需要

发表回复

您的电子邮箱地址不会被公开。

关注全栈程序员社区公众号