kafka学习(五)Spring Boot 整合 Kafka

一、创建Spring boot 工程

创建过程不再描述,创建后的工程结构如下:

POM文件中要加入几个依赖:

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.1.9.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.zhbf</groupId>
    <artifactId>springboot</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>springboot</name>
    <description>Demo project for Spring Boot</description>

    <properties>
        <java.version>1.8</java.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <!--引入kafka依赖-->
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>
        <!-- 添加 gson 依赖 -->
        <dependency>
            <groupId>com.google.code.gson</groupId>
            <artifactId>gson</artifactId>
            <version>2.8.5</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

</project>

启动SpringbootApplication.java,出现下图界面则说明工程创建好了:

二、创建kafka生产者类,并通过控制器调用

kafka生产者类

/**
 * Kafka消息生产类
 */
@Log
@Component
public class KafkaProducer {

    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    @Value("${kafka.topic.user}")
    private String topicUser;//topic名称

    /**
     * 发送用户消息
     *
     * @param user 用户信息
     */
    public void sendUserMessage(User user) {
        GsonBuilder builder = new GsonBuilder();
        builder.setPrettyPrinting();
        builder.setDateFormat("yyyy-MM-dd HH:mm:ss");
        String message = builder.create().toJson(user);
        kafkaTemplate.send(topicUser, message);
        log.info("\n生产消息至Kafka\n" + message);
    }
}

application.yml配置文件

启动ZK、kafka通讯的服务器broker,并启动消费者监听

  启动方式参考上一篇文章,戳这里~

配置一个控制器,即调用kafka生成消息的入口

/**
 * 测试控制器
 * PS:@RestController 注解: 该注解是 @Controller 和 @ResponseBody 注解的合体版
 */
@RestController
@RequestMapping("/kafka")
public class KafkaController {

    @Autowired
    private User user;

    @Autowired
    private KafkaProducer kafkaProducer;

    @RequestMapping("/createMsg")
    public void createMsg() {
        kafkaProducer.sendUserMessage(user);
    }
}

启动SpringbootApplication,并通过浏览器访问控制器,生成消息

可以看到控制台和消费者窗口都打印了kafka生成的消息。

三、创建kafka消费者类,并通过控制器调用

kafka消费者类

@Log
@Component
public class KafkaConsumerDemo {

    public void consume() {
        Properties props = new Properties();

        // 必须设置的属性
        props.put("bootstrap.servers", "127.0.0.1:9092");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("group.id", "group-user");

        // 可选设置属性
        props.put("enable.auto.commit", "true");
        // 自动提交offset,每1s提交一次
        props.put("auto.commit.interval.ms", "1000");
        props.put("auto.offset.reset", "earliest ");
        props.put("client.id", "zy_client_id");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        // 订阅test1 topic
        consumer.subscribe(Collections.singletonList("topic-user"));

        while (true) {
            //  从服务器开始拉取数据
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            records.forEach(record -> {
                System.out.printf("成功消费消息:topic = %s ,partition = %d,offset = %d, key = %s, value = %s%n", record.topic(), record.partition(),
                        record.offset(), record.key(), record.value());
            });
        }
    }
}

重启SpringbootApplication,并通过浏览器访问控制器,消费消息

好文推荐:如何优雅的使用kafka consumer

原文地址:https://www.cnblogs.com/riches/p/11720068.html

时间: 2024-10-08 00:33:22

kafka学习(五)Spring Boot 整合 Kafka的相关文章

Kafka学习--spring boot 整合kafka

一.启动kafka 启动kafka之前一定要启动zookeeper,因为要使用kafka必须要使用zookeeper. windows环境下启动,直接使用kafka自带的zookeeper: E:\kafka_2.12-2.4.0\bin\windows  zookeeper-server-start.bat ..\..\config\zookeeper.properties 接下来启动kafka E:\kafka_2.12-2.4.0\bin\windows  kafka-server-sta

spring boot 整合kafka 报错 Exception thrown when sending a message with key=&#39;null&#39; and payload=JSON to topic proccess_trading_end: TimeoutException: Failed to update metadata after 60000 ms.

org.springframework.kafka.support.LoggingProducerListener- Exception thrown when sending a message with key='null' and payload='{"dataDts":["20180329","20180328","20180327","20180326","20180323"]

spring boot整合kafka

最近项目需求用到了kafka信息中间件,在此做一次简单的记录,方便以后其它项目用到. 引入依赖 <dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency> 配置文件 kafka.consumer.servers=127.0.0.1:9092 kafka.consumer.ena

Spring Kafka和Spring Boot整合实现消息发送与消费简单案例

本文主要分享下Spring Boot和Spring Kafka如何配置整合,实现发送和接收来自Spring Kafka的消息. 先前我已经分享了Kafka的基本介绍与集群环境搭建方法.关于Kafka的介绍请阅读Apache Kafka简介与安装(一),关于Kafka安装请阅读Apache Kafka安装,关于Kafka集群环境搭建请阅读Apache Kafka集群环境搭建 .这里关于服务器环境搭建不在赘述. Spring Kafka整合Spring Boot创建生产者客户端案例 创建一个kafk

Kafka 入门和 Spring Boot 集成

Kafka 入门和 Spring Boot 集成 标签:博客 [TOC] 概述 kafka 是一个高性能的消息队列,也是一个分布式流处理平台(这里的流指的是数据流).由java 和 Scala 语言编写,最早由 LinkedIn 开发,并 2011年开源,现在由 Apache 开发维护. 应用场景 下面列举了一些kafka常见的应用场景. 消息队列 : Kafka 可以作为消息队列使用,可用于系统内异步解耦,流量削峰等场景. 应用监控:利用 Kafka 采集应用程序和服务器健康相关的指标,如应用

Spring Boot 之 Kafka

Kafka是一个分布式的流平台,具有以下功能: 1.发布和订阅消息. 2.以容错的方式存储消息流. 3.实时的流处理. 主要的两大应用: 1.作为实时的流数据通道,在应用程序之间传递消息. 2.构建实时流处理,对数据流进行转换或反应. 四个核心的API: Producer API(输出向),Consumer API(输入向),Streams API(输入&&输出向),Connector API(输入||输出向) https://www.orchome.com/5 Kafka的安装: htt

spring boot整合jsp的那些坑(spring boot 学习笔记之三)

Spring Boot 整合 Jsp 步骤: 1.新建一个spring boot项目 2.修改pom文件 <dependency>            <groupId>org.springframework.boot</groupId>            <artifactId>spring-boot-starter-web</artifactId>        </dependency>        <depend

五、spring boot整合mybatis-plus

spring boot整合mybatis-plus 简介 mybatis 增强工具包,简化 CRUD 操作. 文档 http://mp.baomidou.com http://mybatis.plus 优点 | Advantages 无侵入:Mybatis-Plus 在 Mybatis 的基础上进行扩展,只做增强不做改变,引入 Mybatis-Plus 不会对您现有的 Mybatis 构架产生任何影响,而且 MP 支持所有 Mybatis 原生的特性 依赖少:仅仅依赖 Mybatis 以及 My

activeMQ入门+spring boot整合activeMQ

最近想要学习MOM(消息中间件:Message Oriented Middleware),就从比较基础的activeMQ学起,rabbitMQ.zeroMQ.rocketMQ.Kafka等后续再去学习. 上面说activeMQ是一种消息中间件,可是为什么要使用activeMQ? 在没有使用JMS的时候,很多应用会出现同步通信(客户端发起请求后需要等待服务端返回结果才能继续执行).客户端服务端耦合.单一点对点(P2P)通信的问题,JMS可以通过面向消息中间件的方式很好的解决了上面的问题. JMS规