Spring Boot 使用kafka demo

分手后的思念是犯贱 2021-09-21 07:18 241阅读 0赞

### Spring Boot 使用kafka demo ###

*   *  kafka环境搭建
     *  配置文件配置kafka服务
     *  用到的依赖
     *  创建消息生产者
     *  创建消息服务者

## kafka环境搭建 ##

自行百度资源或[个人另一篇文章](https://blog.csdn.net/qq_22903677/article/details/105906632)

## 配置文件配置kafka服务 ##

spring:
      kafka:
        listener:
          missing-topics-fatal: false # 防止topic 不存在
        bootstrap-servers:
          - 192.168.8.108:9082  #本地kafka 服务
    server:
      port: 7800

## 用到的依赖 ##

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

## 创建消息生产者 ##

这里为了简单测试使用KafkaTemplate 直接注入模版

@RestController
    @RequestMapping
    public class TestController { 
    
        @Resource
        private KafkaTemplate kafkaTemplate;
    
        @ApiOperation("数据接入kafka topic")
        @PostMapping("/topic/{topicName}")
        public void dataAccessTopic(@PathVariable @ApiParam(value = "kafka topic名称,必须已注册", required = true) String topicName,
                                                   @RequestBody String payload) { 
                kafkaTemplate.send(topicName, payload);
        }
    
    
    }

## 创建消息服务者 ##

服务消费采用注解@EnableKafka与@KafkaListener

/**
     * 监听kafka 消息队列
     */
    @Slf4j
    @Component
    @EnableKafka
    public class KafkaConsumerListenser { 
    
    
        /**
         *
         * @param records
         */
        @KafkaListener(groupId = "groupA", topics = "topicA")
        void listener(String records) { 
            List<Map<String, String>> dataList = new ArrayList<Map<String, String>>();
            String tabliName = null;
            log.info(records);
        }
    }

[demo地址][demo]

[demo]: https://github.com/wujiawei1207537021/kafka-demo