原创

SpringBoot进阶教程(六十二)整合Kafka

在上一篇文章《Linux安装Kafka》中,已经介绍了如何在Linux安装Kafka,以及Kafka的启动/关闭和创建发话题并产生消息和消费消息。这篇文章就介绍介绍SpringBoot整合Kafka。

v创建项目

若是已有的项目中添加kafka, 请直接跳至1.3

1.1 创建springboot:

SpringBoot进阶教程(六十二)整合Kafka

SpringBoot进阶教程(六十二)整合Kafka

1.2 选web和kafka:

SpringBoot进阶教程(六十二)整合Kafka

SpringBoot进阶教程(六十二)整合Kafka

1.3 已有的项目中添加kafka, pom.xml中添加依赖

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

1.4 整体架构目录:

SpringBoot进阶教程(六十二)整合Kafka

v配置项目

2.1 设置application.yml

spring:
  kafka:
    bootstrap-servers: 127.0.0.1:9092 #指定kafka server的地址,集群配多个,中间,逗号隔开
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer:
      group-id: default_consumer_group #群组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
server:
  port: 8500

项目默认生成的是applicaiton.properties,直接重命名修改文件后缀名为yml即可。

2.2 添加生产者ProducerController

package com.toutou.Controller;

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;

/**
 * @author toutou
 * @date by 2019/08
 */
@RestController
public class ProducerController {
    @Autowired
    private KafkaTemplate<String,Object> kafkaTemplate;
    @RequestMapping("message/send")
    public String send(String msg){
        kafkaTemplate.send("demo", msg); //使用kafka模板发送信息
        return "success";
    }
}

2.3 添加消费者ConsumerDemo

package com.toutou.Consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

/**
 * @author toutou
 * 监听服务器上的kafka是否有相关的消息发过来
 * @date by 2019/08
 */
@Component
public class ConsumerDemo {
    /**
     * 定义此消费者接收topics = "demo"的消息,与controller中的topic对应上即可
     * @param record 变量代表消息本身,可以通过ConsumerRecord<?,?>类型的record变量来打印接收的消息的各种信息
     */
    @KafkaListener(topics = "demo")
    public void listen (ConsumerRecord<?, ?> record){
        System.out.printf("topic is %s, offset is %d, value is %s \n", record.topic(), record.offset(), record.value());
    }
}

v启动测试

3.1 测试生产者

SpringBoot进阶教程(六十二)整合Kafka

3.2 消费者效果

SpringBoot进阶教程(六十二)整合Kafka

v源码地址

https://github.com/toutouge/javademosecond/tree/master/hellokafka


作  者:请叫我头头哥
出  处:http://www.cnblogs.com/toutou/
关于作者:专注于基础平台的项目开发。如有问题或建议,请多多赐教!
版权声明:本文版权归作者和博客园共有,欢迎转载,但未经作者同意必须保留此段声明,且在文章页面明显位置给出原文链接。
特此声明:所有评论和私信都会在第一时间回复。也欢迎园子的大大们指正错误,共同进步。或者直接私信
声援博主:如果您觉得文章对您有帮助,可以点击文章右下角推荐一下。您的鼓励是作者坚持原创和持续写作的最大动力!

正文到此结束
本文目录