Springboot+Kafka最快速入门

网络安全邮递员 lv.1

发布时间：2022-12-01 22:17:10 465

相关标签： # html# spring# scala# apache# 信息

前言

当应用系统达到一定规模后，随着访问量的增加，随之而来的是，对系统的性能要求也日益增加，这时我们就需要运用一些第三方工具组件来提高性能，提高性能的主要手段之一是使用消息中间件，因为它的异步处理机制能大大提高系统吞吐量，另外，使用消息中间件后，其天然具有的解耦效果同时也便于我们更有效的组织规划系统功能。消息中间件有很多种，常见的有ActiveMQ，RabbitMQ，RocketMQ，Kafka等。网上有很多比较他们之间异同的文章，可自行阅读, 但它们的主要功能是一致的。这篇文章里，我们试图用尽量简洁的文字描述一下怎样最快速地入手其中之一——Kafka。

安装Kafka运行环境

安装启动Zookeeper

Kafka和Zookeeper都是Apache的子项目，zookeeper 主要是用来解决分布式应用中经常遇到的一些数据管理问题，如：统一命名服务、状态同步服务、集群管理、分布式应用配置项的管理等。

Kafka必须和Zookeeper一起使用，先到apache网站下载zookerper的发行版本，我目前看到的最新稳定版是3.7.1

https://zookeeper.apache.org/releases.html

Springboot+Kafka最快速入门_kafka

下载后，解压到一个目录，比如D:\apache-zookeeper-3.8.0-bin

如果你是在windows下，直接运行bin目录下的zkServer.cmd文件即可启动。

安装Scalar

Kafka是用Scala语言写的，必须要先安装Scala运行环境才能使用，在这里下载发行版本。

https://www.scala-lang.org/download/

这里我们主要是学习，下载安装Windows版本即可

Springboot+Kafka最快速入门_java_02

安装启动Kafka

在Apache网站下载Kafka,下载发行版，截止此文，目前的最新版本为3.3.1

https://kafka.apache.org/downloads.html

Springboot+Kafka最快速入门_zookeeper_03

解压到一个目录，比如D:\kafka_2.13-3.3.1

如果是在Windows下编写一个批处理文件kalfka.cmd

内容如下：

@echo off

@echo off
set KAFKA_HOME=D:\kafka_2.13-3.3.1
%KAFKA_HOME%\bin\windows\kafka-server-start %KAFKA_HOME%\config\server.properties

KAFKA_HOME=后面改为你自己的路径。保存后，双击运行我们就启动了kafka.

注意，不能直接双击bin目录下的kafka-server-start.sh运行kafka，否则会报错。

到此为止我们就建好了Kafka的运行环境。

Linux环境下的搭建比较类似，不做赘述。

SpringBoot里使用Kafka

SpringBoot Starter向导

这里我们以开发环境Eclipse为例，说明怎样引入Kafka（其它IDE 也是类似，大同小异），

当然，没有IDE也是可以的，只是要手工引入库文件，稍微麻烦一点。

选择菜单New/Project，对话框中选择 Spring Boot下的Spring Starter Project

Springboot+Kafka最快速入门_入门_04

输入相关的项目信息。注意JDK版本与你的实际情况相符。

Springboot+Kafka最快速入门_java_05

在选择依赖库界面，注意Spring Boot版本需要与你的JDK版本匹配，如果你选择了SpringBoot 3.x.x, 那你的JDK必须是17以上，后面的tomcat也要使用tomcat10, 而tomcat10与tomcat9区别比较大。我是用的JDK8,所以这里只能选择SpringBoot 2.x.x, 勾选上Web和2个Kafka依赖库，

Springboot+Kafka最快速入门_入门_06

最后点击Finish就生成了项目骨架了，现在我们加一点代码上去。

加入Kafka相关代码

首先我们加入Producer

package com.example.kafka;

import java.util.Map;

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.RequestParam;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author chenyh-a
 * 
 */
@RestController
public class OrderController {
  @Autowired
  private KafkaTemplate<String, String> kt;

  @RequestMapping(value = "/send")
  public String send(@RequestParam Map<String, String> params) {
    // change it to actual code, collect front-end data,etc.
    String sorder = "{\"orderId\":12345}";
    kt.send("TOPIC_ORDER", "key", sorder);
    return "{\"result\":\"SUCCESS\"}";
  }
}

因为是Web应用，这里我们说的Producer 就是一个Web应用的一个Controller而已。中间的变量sorder就是我们要发送的数据信息，可以改为收集从前台界面传来的内容。

下面是Consumer示例：

package com.example.kafka;

import org.springframework.stereotype.Component;
import org.springframework.kafka.annotation.KafkaListener;

/**
 * @author chenyh-a
 * 
 */
@Component
public class OrderConsumer {
  @KafkaListener(topics = { "TOPIC_ORDER" }, groupId = "GROUP_LOGISTICS")
  public void onMessage(String message) {
    System.out.println("Order consumer received:" + message);
    //write actual consumer code here
  }
}

Consumer是一个单独的类，简单展示了怎么接到Producer发送的数据。注意KafkaListner的参数说明。topic必须与Producer发送时的topic保持一致。在实际应用中，Consumer一般是分开的，不会跟Producer应用放一起。

有了基本的Producer和Consumer，我们就可以运行观察Kafka功能了。

在运行之前，需要先把tomcat默认的端口号8080改成其它，比如8085，因为8080已经被zookeeper占用了。

Springboot+Kafka最快速入门_zookeeper_07

好了，现在可以发布运行了，发布完在浏览器中访问

http://localhost:8085/kftest/send

即可看到结果

Springboot+Kafka最快速入门_zookeeper_08

同时后台也可以看到Consumer打印出的内容。

Springboot+Kafka最快速入门_zookeeper_09

需要注意的是，springboot / jdk / tomcat这3者的版本必须匹配，否则会出错。这里我使用的jdk8, springboot 2.x.x, tomcat 9.x.x , 如果选择spring3.x.x，你的jdk 版本需要是jdk17, WEB容器必须是tomcat 10.x.x。

到此为止，有关Kafka的基本示例我们就展示完毕。当然这只是一个类似 Hello world 的最简单示例，进阶内容需要继续阅读相关的文档。

示例源代码

https://github.com/chenyh-a/kftest

文章来源： https://blog.51cto.com/u_15817067/5889187

特别声明：以上内容（图片及文字）均为互联网收集或者用户上传发布，本站仅提供信息存储服务！如有侵权或有涉及法律问题请联系我们。