消息中间件系列教程(22) -Kafka- SpringBoot集成Kafka

简介: 消息中间件系列教程(22) -Kafka- SpringBoot集成Kafka

引言

代码已提交至Github,有兴趣的同学可以下载来看看:https://github.com/ylw-github/SpringBoot-Kafka-Demo

搭建教程上两篇博客又讲,可以参考:

为了方便起见,建议参考: 《消息中间件系列教程(21) -Kafka- 集群搭建(自带Zookeeper)》

SpringBoot集成Kafka

1.新建Maven项目

2.添加Maven依赖:

<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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <groupId>ylw</groupId>
  <artifactId>com.ylw.springboot.kafka</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  <parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.0.1.RELEASE</version>
  </parent>
  <dependencies>
    <!-- springBoot集成kafka -->
    <dependency>
      <groupId>org.springframework.kafka</groupId>
      <artifactId>spring-kafka</artifactId>
    </dependency>
    <!-- SpringBoot整合Web组件 -->
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
  </dependencies>
</project>

3.添加application.yml

# kafka
spring:
  kafka:
    # kafka服务器地址(可以多个)
    bootstrap-servers: 192.168.162.131:9092,192.168.162.132:9092,192.168.162.133:9092
    consumer:
      # 指定一个默认的组名
      group-id: kafka2
      # earliest:当各分区下有已提交的offset时,从提交的offset开始消费;无提交的offset时,从头开始消费
      # latest:当各分区下有已提交的offset时,从提交的offset开始消费;无提交的offset时,消费新产生的该分区下的数据
      # none:topic各分区都存在已提交的offset时,从offset后开始消费;只要有一个分区不存在已提交的offset,则抛出异常
      auto-offset-reset: earliest
      # key/value的反序列化
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    producer:
      # key/value的序列化
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      # 批量抓取
      batch-size: 65536
      # 缓存容量
      buffer-memory: 524288
      # 服务器地址
      bootstrap-servers: 192.168.162.131:9092,192.168.162.132:9092,192.168.162.133:9092

4.Controller代码:

package com.ylw;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
public class KafkaController {
  /**
   * 注入kafkaTemplate
   */
  @Autowired
  private KafkaTemplate<String, String> kafkaTemplate;
  /**
   * 发送消息的方法
   *
   * @param key
   *            推送数据的key
   * @param data
   *            推送数据的data
   */
  private void send(String key, String data) {
    // topic 名称 key data 消息数据
    kafkaTemplate.send("my_test", key, data);
  }
  // test 主题 1 my_test 3
  @RequestMapping("/kafka")
  public String testKafka() {
    int iMax = 6;
    for (int i = 1; i < iMax; i++) {
      send("key" + i, "data" + i);
    }
    return "success";
  }
  /**
   * 消费者使用日志打印消息
   */
  @KafkaListener(topics = "my_test")
  public void receive(ConsumerRecord<?, ?> consumer) {
    System.out.println("topic名称:" + consumer.topic() + ",key:" + consumer.key() + ",分区位置:" + consumer.partition()
        + ", 下标" + consumer.offset());
  }
}

5.启动类:

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

6.启动程序,浏览器请求:http://localhost:8080/kafka

控制台收到消息,(消费者消费消息):

使用Zookeeper可视化工具可以看到新增了一个my_test主题:

本文完!

目录
相关文章
|
6天前
|
消息中间件 Java Kafka
Springboot集成高低版本kafka
Springboot集成高低版本kafka
|
6天前
|
机器学习/深度学习 Python
CatBoost高级教程:深度集成与迁移学习
CatBoost高级教程:深度集成与迁移学习【2月更文挑战第17天】
30 1
|
6天前
|
机器学习/深度学习 算法 Python
CatBoost中级教程:集成学习与模型融合
CatBoost中级教程:集成学习与模型融合【2月更文挑战第13天】
51 3
|
6天前
|
机器学习/深度学习 算法 Python
LightGBM高级教程:深度集成与迁移学习
LightGBM高级教程:深度集成与迁移学习【2月更文挑战第6天】
116 4
|
6天前
|
Kubernetes Java 容器
部署 Spring Boot 应用到 K8S 教程
部署 Spring Boot 应用到 K8S 教程
71 0
|
6天前
|
消息中间件 Kafka Apache
Flink 提供了与 Kafka 集成的官方 Connector,使得 Flink 能够消费 Kafka 数据
【2月更文挑战第6天】Flink 提供了与 Kafka 集成的官方 Connector,使得 Flink 能够消费 Kafka 数据
75 2
|
6天前
|
Java 关系型数据库 MySQL
保姆级教程——将springboot项目部署到阿里云服务器包含环境配置(小白包会)
本文档详细介绍了将SpringBoot项目部署到阿里云服务器的步骤。首先,通过Xshell连接服务器,使用公网IP地址。接着,下载JDK的Linux版本,使用XFTP上传并解压,配置环境变量。然后,安装MySQL 5.7,包括下载YUM源、安装、启动服务以及修改root密码和开启远程访问。最后,将SpringBoot项目打包成jar,上传至服务器,使用`java -jar`命令运行,通过`nohup`确保服务持续运行。配置安全组以允许远程访问。
|
6天前
|
消息中间件 Java Linux
RabbitMQ教程:Linux下安装、基本命令与Spring Boot集成
RabbitMQ教程:Linux下安装、基本命令与Spring Boot集成
|
6天前
|
jenkins Java 持续交付
Docker搭建持续集成平台Jenkins最简教程
Jenkins 是一个广泛使用的开源持续集成工具,它能够自动化构建、测试和部署软件项目。在本文中,我们将使用 Docker 搭建一个基于 Jenkins 的持续集成平台。
147 2
|
6天前
|
消息中间件 存储 物联网

热门文章

最新文章