Kafka（五）springboot集成kafka

一、添加kafka依赖

在https://mvnrepository.com/中搜索kafka，我们使用spring本身支持的依赖spring-kafka。

这里面需要注意点版本的问题，参考下图做对比，不同版本的springboot支持不同版本的kafka：

版本对应关系

也可以自己去官方看这个对应关系：https://spring.io/projects/spring-kafka/

对比上图去maven仓库找相应版本就可以了：https://mvnrepository.com/

maven

我的springboot版本是2.2.5的，所以使用kafka的版本如下所示：

<!-- https://mvnrepository.com/artifact/org.springframework.kafka/spring-kafka -->
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
    <version>2.3.5.RELEASE</version>
</dependency>

如果版本不对应会有很多的问题，通常kafka的版本要比springboot高一个版本就行了，比如2.2.5则kafka是2.3.5这样。

二、添加配置文件

我这里使用一个共工程模拟生产者和消费者。
我的springboot工程使用的配置文件是yaml类型的，所以按照如下方式添加：

spring:
  kafka:
    bootstrap-servers: 192.168.184.134:9092,192.168.184.135:9092,192.168.184.136:9092
    producer:
      # 发生错误后，消息重发的次数。
      retries: 0
      # 键的序列化方式
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      # 值的序列化方式
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer:
      group-id: test
      # 键的反序列化方式
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      # 值的反序列化方式
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    listener:
      # 但消费者监听的topic不存在时，保证能够是项目启动
      missing-topics-fatal: false

配置还有很多，我们这里先集成，后面会逐渐增加配置。

三、代码实现

3.1生产者

package com.cloud.bssp.message.kafka.producer;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

/**
 * kafka生产者
 *
 * @author weirx
 * @date 2021/02/03 14:22
 **/
@Component
public class KafkaProducer {

    @Autowired
    private KafkaTemplate kafkaTemplate;

    /**
     * kafka消息发送
     * @param
     * @author weirx
     * @return void
     * @date: 2021/2/3
     */
    public void send(String topic){
        kafkaTemplate.send(topic,"hello kafka");
    }
}

3.2消费者

package com.cloud.bssp.message.kafka.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Optional;

/**
 * kafka消费者
 *
 * @author weirx
 * @date 2021/02/03 15:01
 **/
@Slf4j
@Component
public class KafkaConsumer {

    @KafkaListener(topics = {"test-kafka"})
    public void consumer(ConsumerRecord<?, ?> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            Object message = kafkaMessage.get();
            log.info("----------------- record =" + record);
            log.info("------------------ message =" + message);
        }
    }
}

3.3测试类

package com.cloud.bssp.message.kafka;

import com.cloud.bssp.message.kafka.producer.KafkaProducer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * kafka测试类
 *
 * @author weirx
 * @date 2021/02/03 15:03
 **/
@RestController
@RequestMapping("test/kafka")
public class TestKafka {

    @Autowired
    private KafkaProducer producer;

    @RequestMapping("/send/{topic}")
    public void send(@PathVariable String topic){
        producer.send(topic);
    }
}

四、测试

这里我将topic作为了参数进行传递，方便我们观察。
我们执行以下测试接口：

http://localhost:8085/test/kafka/send/test-kafka

看下控制台，消费者输出的结果：

2021-02-03 17:12:44.654  INFO 5080 --- [ntainer#0-0-C-1] c.c.b.m.kafka.consumer.KafkaConsumer     : ----------------- record =ConsumerRecord(topic = test-kafka, partition = 5, leaderEpoch = 0, offset = 0, CreateTime = 1612343564622, serialized key size = -1, serialized value size = 11, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = hello kafka)
2021-02-03 17:12:44.654  INFO 5080 --- [ntainer#0-0-C-1] c.c.b.m.kafka.consumer.KafkaConsumer     : ------------------ message =hello kafka

从上结果看到：
topic：test-kafka
partition：5
offset：0

我们从kafkatool中看看结果，在分区5中增加了一条消息：

kafkatool

这里为什么有10个分区呢？在上一章节我们搭建kafka集群的时候，默认给了10个partition，关于partition的作用，在我们学会使用后，后面的章节会详细介绍。

offset是当前partition中的数据个数的偏移量，从0开始。

下面我们再次发送10条消息，使消息再次发送到partition5中。结果如下：

2021-02-03 17:15:17.129  INFO 5080 --- [ntainer#0-0-C-1] c.c.b.m.kafka.consumer.KafkaConsumer     : ----------------- record =ConsumerRecord(topic = test-kafka, partition = 9, leaderEpoch = 0, offset = 0, CreateTime = 1612343717117, serialized key size = -1, serialized value size = 11, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = hello kafka)
2021-02-03 17:15:17.129  INFO 5080 --- [ntainer#0-0-C-1] c.c.b.m.kafka.consumer.KafkaConsumer     : ------------------ message =hello kafka
2021-02-03 17:20:45.456  INFO 5080 --- [ntainer#0-0-C-1] c.c.b.m.kafka.consumer.KafkaConsumer     : ----------------- record =ConsumerRecord(topic = test-kafka, partition = 0, leaderEpoch = 0, offset = 0, CreateTime = 1612344045453, serialized key size = -1, serialized value size = 11, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = hello kafka)
2021-02-03 17:20:45.456  INFO 5080 --- [ntainer#0-0-C-1] c.c.b.m.kafka.consumer.KafkaConsumer     : ------------------ message =hello kafka
2021-02-03 17:21:12.999  INFO 5080 --- [ntainer#0-0-C-1] c.c.b.m.kafka.consumer.KafkaConsumer     : ----------------- record =ConsumerRecord(topic = test-kafka, partition = 4, leaderEpoch = 0, offset = 0, CreateTime = 1612344072988, serialized key size = -1, serialized value size = 11, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = hello kafka)
2021-02-03 17:21:12.999  INFO 5080 --- [ntainer#0-0-C-1] c.c.b.m.kafka.consumer.KafkaConsumer     : ------------------ message =hello kafka
2021-02-03 17:21:15.983  INFO 5080 --- [ntainer#0-0-C-1] c.c.b.m.kafka.consumer.KafkaConsumer     : ----------------- record =ConsumerRecord(topic = test-kafka, partition = 3, leaderEpoch = 0, offset = 0, CreateTime = 1612344075980, serialized key size = -1, serialized value size = 11, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = hello kafka)
2021-02-03 17:21:15.983  INFO 5080 --- [ntainer#0-0-C-1] c.c.b.m.kafka.consumer.KafkaConsumer     : ------------------ message =hello kafka
2021-02-03 17:21:16.799  INFO 5080 --- [ntainer#0-0-C-1] c.c.b.m.kafka.consumer.KafkaConsumer     : ----------------- record =ConsumerRecord(topic = test-kafka, partition = 2, leaderEpoch = 0, offset = 0, CreateTime = 1612344076796, serialized key size = -1, serialized value size = 11, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = hello kafka)
2021-02-03 17:21:16.799  INFO 5080 --- [ntainer#0-0-C-1] c.c.b.m.kafka.consumer.KafkaConsumer     : ------------------ message =hello kafka
2021-02-03 17:21:17.614  INFO 5080 --- [ntainer#0-0-C-1] c.c.b.m.kafka.consumer.KafkaConsumer     : ----------------- record =ConsumerRecord(topic = test-kafka, partition = 1, leaderEpoch = 0, offset = 0, CreateTime = 1612344077612, serialized key size = -1, serialized value size = 11, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = hello kafka)
2021-02-03 17:21:17.614  INFO 5080 --- [ntainer#0-0-C-1] c.c.b.m.kafka.consumer.KafkaConsumer     : ------------------ message =hello kafka
2021-02-03 17:21:18.487  INFO 5080 --- [ntainer#0-0-C-1] c.c.b.m.kafka.consumer.KafkaConsumer     : ----------------- record =ConsumerRecord(topic = test-kafka, partition = 8, leaderEpoch = 0, offset = 0, CreateTime = 1612344078484, serialized key size = -1, serialized value size = 11, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = hello kafka)
2021-02-03 17:21:18.487  INFO 5080 --- [ntainer#0-0-C-1] c.c.b.m.kafka.consumer.KafkaConsumer     : ------------------ message =hello kafka
2021-02-03 17:21:19.527  INFO 5080 --- [ntainer#0-0-C-1] c.c.b.m.kafka.consumer.KafkaConsumer     : ----------------- record =ConsumerRecord(topic = test-kafka, partition = 7, leaderEpoch = 0, offset = 0, CreateTime = 1612344079524, serialized key size = -1, serialized value size = 11, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = hello kafka)
2021-02-03 17:21:19.527  INFO 5080 --- [ntainer#0-0-C-1] c.c.b.m.kafka.consumer.KafkaConsumer     : ------------------ message =hello kafka
2021-02-03 17:21:20.505  INFO 5080 --- [ntainer#0-0-C-1] c.c.b.m.kafka.consumer.KafkaConsumer     : ----------------- record =ConsumerRecord(topic = test-kafka, partition = 6, leaderEpoch = 0, offset = 0, CreateTime = 1612344080500, serialized key size = -1, serialized value size = 11, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = hello kafka)
2021-02-03 17:21:20.505  INFO 5080 --- [ntainer#0-0-C-1] c.c.b.m.kafka.consumer.KafkaConsumer     : ------------------ message =hello kafka
2021-02-03 17:21:21.440  INFO 5080 --- [ntainer#0-0-C-1] c.c.b.m.kafka.consumer.KafkaConsumer     : ----------------- record =ConsumerRecord(topic = test-kafka, partition = 5, leaderEpoch = 0, offset = 1, CreateTime = 1612344081437, serialized key size = -1, serialized value size = 11, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = hello kafka)
2021-02-03 17:21:21.440  INFO 5080 --- [ntainer#0-0-C-1] c.c.b.m.kafka.consumer.KafkaConsumer     : ------------------ message =hello kafka

如上结果，又发送了10条消息，消息随机的负载到每个partition中，第10条再次记录到了partition 5中，其offset加1了。

kafkatool

到此为止，已经完成了springboot和kafka的集成了，后面会逐渐深入到每个特性和组件。

©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 220,002评论 6赞 509
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 93,777评论 3赞 396
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 166,341评论 0赞 357
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 59,085评论 1赞 295
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 68,110评论 6赞 395
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 51,868评论 1赞 308
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 40,528评论 3赞 420
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 39,422评论 0赞 276
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 45,938评论 1赞 319
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 38,067评论 3赞 340
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 40,199评论 1赞 352
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 35,877评论 5赞 347
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 41,540评论 3赞 331
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 32,079评论 0赞 23
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 33,192评论 1赞 272
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 48,514评论 3赞 375
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 45,190评论 2赞 357

推荐阅读更多精彩内容