在Spring Boot应用集成Kafka读写消息

ゞ浴缸里的玫瑰 2022-09-05 01:40 136阅读 0赞

### 文章目录 ###

*  在Spring Boot应用集成Kafka读写消息
 *   *  前言
     *  在本地以Docker方式安装和运行Kafka
     *  Spring Boot应用
     *   *  Spring Kafka依赖
         *  应用的Kafka配置
         *  Producer类
         *  Consumer类
         *  Controller类
         *  Constants类
     *  测试发送与接收消息
     *  小结
     *  参考文档
     *  扩展阅读

# 在Spring Boot应用集成Kafka读写消息 #

## 前言 ##

本文描述了一个在Spring Boot应用中集成Kafka读写消息的例子。

为简单起见，Producer和Consumer都在同一个应用中。

用例：

*  用户调用REST API，发送一个消息。
 *  Spring Boot Controller调用Producer，将该消息写入`users`的Kafka topic中。
 *  Consumer订阅（侦听）该`users` 的Kafka topic，并从该topic中读取消息。

环境和工具：

*  Spring Boot 2.4.0
 *  Java 1.8
 *  Kafka 2.6.0
 *  Lombok - 用来简化代码

## 在本地以Docker方式安装和运行Kafka ##

cd scripts/kafka
    
    # start
    docker-compose up -d 
    
    # stop
    docker-compose stop

参见：

*  [在本地以Docker方式安装和运行Kafka][Docker_Kafka]

## Spring Boot应用 ##

### Spring Kafka依赖 ###

在Spring Boot项目中引入Spring Kafka依赖：

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

### 应用的Kafka配置 ###

在Spring Boot应用的`application.yaml`中配置：

server:
      port: 8080
    spring:
      kafka:
        consumer:
          bootstrap-servers: localhost:9092
          group-id: demo-group
          auto-offset-reset: earliest
          key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
        producer:
          bootstrap-servers: localhost:9092
          key-serializer: org.apache.kafka.common.serialization.StringSerializer
          value-serializer: org.apache.kafka.common.serialization.StringSerializer

说明：

*  本地Kafka server的地址为 `localhost:9092`
 *  Consumer Group ID为`demo-group`

### Producer类 ###

@Service
    @RequiredArgsConstructor
    @Slf4j
    public class Producer { 
        private final KafkaTemplate<String, String> kafkaTemplate;
    
        public void sendMessage(String message) { 
            log.info(String.format("#### -> Producing message -> %s", message));
            kafkaTemplate.send(KafkaConstants.TOPIC_GREETING, message);
        }
    }

说明：

*  在Producer中注入和调用`KafkaTemplate`来发送消息。
 *  这里用Lombok的`@RequiredArgsConstructor`注解和`@Slf4j`注解来使用constructor-inject方式来注入bean，并引入logger。

### Consumer类 ###

@Service
    @Slf4j
    public class Consumer { 
    
        @KafkaListener(topics = KafkaConstants.TOPIC_GREETING, groupId = KafkaConstants.CONSUMER_GROUP)
        public void consume(String message) { 
            log.info(String.format("#### -> Consumed message -> %s", message));
        }
    }

说明：

*  在Consumer中用`@KafkaListener`注解来配置订阅的Kafka topic和所属的Consumer Group。
 *  这里用Lombok的`@Slf4j`注解来引入logger。

### Controller类 ###

定义一个`KafkaController`类来供外部调用API来触发Producer往Kafka topic中发送消息。

@RestController
    @RequestMapping("/kafka")
    @RequiredArgsConstructor
    public class KafkaController { 
    
        private final Producer producer;
    
        @PostMapping("/publish")
        public void sendMessage(@RequestParam("message") String message) { 
            producer.sendMessage(message);
        }
    }

说明：

*  这里用Lombok的`@RequiredArgsConstructor`注解来使用constructor-inject方式来注入bean。

### Constants类 ###

定义一个`KafkaConstants` 类来统一定义应用中用到的Kafka常量。

@UtilityClass
    public class KafkaConstants { 
        // Topics
        public static final String TOPIC_GREETING = "greeting";
    
        // Consumer Group
        public static final String CONSUMER_GROUP = "demo-group";
    
    }

说明：

*  这里用Lombok的`@UtilityClass`注解防止该工具类被继承和被实例化。

## 测试发送与接收消息 ##

调用API发送消息：

curl -X POST http://localhost:8080/kafka/publish -F 'message=luckyboy'

或者用HttPie工具来测试：

http -f :8080/kafka/publish message="hello"

消息流转过程：API -> Controller -> Producer -> Kafka Topic -> Consumer

在Spring Boot应用的日志中可以查看Producer和Consumer的日志。

## 小结 ##

本文代码示例：

*  [https://github.com/cookcodeblog/spring-kafka-demo][https_github.com_cookcodeblog_spring-kafka-demo]

## 参考文档 ##

*  [How to Work with Apache Kafka in Your Spring Boot Application][]
 *  [HTTPie - API测试工具的另一个选择][HTTPie - API]

## 扩展阅读 ##

*  [Spring Kafka Reference][]
 *  [Kafka Introduction][]

[Docker_Kafka]: https://cookcode.blog.csdn.net/article/details/119728087
[https_github.com_cookcodeblog_spring-kafka-demo]: https://github.com/cookcodeblog/spring-kafka-demo
[How to Work with Apache Kafka in Your Spring Boot Application]: https://www.confluent.io/blog/apache-kafka-spring-boot-application/
[HTTPie - API]: https://cookcode.blog.csdn.net/article/details/117307027
[Spring Kafka Reference]: https://docs.spring.io/spring-kafka/reference/html/
[Kafka Introduction]: https://kafka.apache.org/intro