kafka集成外部系统

1.集成 Springboot

pom：

    4.0.0
    com.atguigu.springboot
    springboot-kafka
    0.0.1-SNAPSHOT
    springboot-kafka
    springboot-kafka
    
        1.8
        UTF-8
        UTF-8
        2.6.13
    
    
        
            org.springframework.boot
            spring-boot-starter-web
        
        
            org.springframework.kafka
            spring-kafka
        
        
            org.projectlombok
            lombok
            true
        
        
            org.springframework.boot
            spring-boot-starter-test
            test
        
        
            org.springframework.kafka
            spring-kafka-test
            test
        
    
    
        
            
                org.springframework.boot
                spring-boot-dependencies
                ${spring-boot.version}
                pom
                import
            
        
    
    
        
            
                org.apache.maven.plugins
                maven-compiler-plugin
                3.8.1
                
                    1.8
                    1.8
                    UTF-8
                
            
            
                org.springframework.boot
                spring-boot-maven-plugin
                ${spring-boot.version}
                
                    com.atguigu.springboot.springbootkafka.SpringbootKafkaApplication
                    true
                
                
                    
                        repackage
                        
                            repackage
                        
                    
                
            
        
    

2.配置

# 应用服务 WEB 访问端口
server.port=8080
# =========消费者配置开始=========
# 指定 kafka 的地址
spring.kafka.bootstrap-servers=hadoop100:9092,hadoop101:9092,hadoop102:9092
# 指定 key 和 value 的反序列化器
spring.kafka.producer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.producer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
#指定消费者组的 group_id
spring.kafka.consumer.group-id=atguigu
# =========消费者配置结束=========
# 指定 key 和 value 的反序列化器
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

3.生产者

package com.atguigu.springboot.springbootkafka.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;
@RestController
public class ProducerController {
    @Autowired
    KafkaTemplate kafka;
    @RequestMapping("/atguigu")
    public String data(String msg) {
        kafka.send("first",msg);
        return "ok";
    }
}

4.消费者

package com.atguigu.springboot.springbootkafka.controller;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.KafkaListener;
@Configuration
public class KafkaConsumer {
    @KafkaListener(topics = "first")
    public void consumeTopic(String msg) {
        System.out.println("收到的信息: " + msg);
    }
}