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)


相关推荐

  • QT——开发入门简介

    QT——开发入门简介1、你了解Qt吗?Qt是一种基于C++的跨平台图形用户界面应用程序开发框架。如何跨平台?上到服务器上位机,下到嵌入式GUI,上天入地无所不能。Qt最早是由1991年由QtCompany开发,但是到2008年,QtCompany科技被诺基亚公司收购,是的,就是拥有着我们很多情怀的诺基亚。但在2012年,Qt又被Digia收购。等到了2014年,跨平台集成开发环境QtCreator3.1….

  • SpringCloud(八)—-OpenFeign通讯优化

    SpringCloud(八)—-OpenFeign通讯优化

    2020年11月12日
  • 转罗马字符_罗马字符是什么意思

    转罗马字符_罗马字符是什么意思publicclassNumToRoman{publicstaticstringToRomanNum(intpNum){string[,]a=newstring[3,10]{{“”,”I”,”II”,”III”,”IV”,”V”,”VI”,”VII”,”VIII”,

  • 神经网络学习 之 BP神经网络

    神经网络学习 之 BP神经网络上一次我们讲了M-P模型,它实际上就是对单个神经元的一种建模,还不足以模拟人脑神经系统的功能。由这些人工神经元构建出来的网络,才能够具有学习、联想、记忆和模式识别的能力。BP网络就是一种简单的人工神经网络。本文具体来介绍一下一种非常常见的神经网络模型——反向传播(BackPropagation)神经网络。概述BP(BackPropagation)神经网络是1986年由Rumelhart和Mc

  • USES_CONVERSION宏定义

    USES_CONVERSION宏定义USES_CONVERSION是用来转换类型的(比如T2A等转换需用此宏),比如我们很常见的问题:在Socket编程时候,我们的IP地址从界面上输进去一般都使用CString类型的,可是在SOCKADDR_IN中的inet_addr却是const char *我们就不能直接用CString来用。我们就可以使用T2A()宏了。 SOCKADDR_IN localaddr; …

  • 随机数生成机制

    随机数生成机制

发表回复

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

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