[Spring Boot]Kafka集成与简单实现

简述

实现Spring Boot集成Kafka，并且简单实现生产者与消费者的场景
有关安装kafka可参照往期文章
[Docker]kafka安装

集成

引入Kafka配置

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

配置文件

server:
  port: 19900
spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      auto-offset-reset: latest
      max-poll-records: 100
      max-partition-fetch-bytes: 1000000

设置Kafka消费者

package com.fir.rbac;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.kafka.annotation.KafkaListener;


@SpringBootApplication
public class Application {


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

    @KafkaListener(id = "webGroup", topics = "topic_input")
    public void listen(String input) {
        System.out.println("input value: " + input);
    }
}

向指定主题发送消息

package com.fir.rbac.controller;

import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;


/**
 * 测试方法
 *
 * @author dpe
 */
@Slf4j
@RestController
public class TestController {


    @Resource
    private KafkaTemplate<Object, Object> template;



    @RequestMapping("/send")
    public void login(String input) {

        this.template.send("topic_input", input);
    }

}

Spring Boot启动应用

启动项目

调用接口发送消息

消费者接受消息