kafka-消息中间键(springboot集成)

发布时间：2023-02-09 文章分类：编程知识 投稿人：赵颖 字号： 默认 | 大 | 超大 打印

特性

追求高吞吐量，适合产生大量数据的互联网服务的数据收集业务

kafka入门

1.导入依赖

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <!-- kafkfa -->
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
        <exclusions>
            <exclusion>
                <groupId>org.apache.kafka</groupId>
                <artifactId>kafka-clients</artifactId>
            </exclusion>
        </exclusions>
    </dependency>
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
    </dependency>
    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>fastjson</artifactId>
    </dependency>
</dependencies>

2.配置文件

生产者：

spring:
  kafka:
    bootstrap-servers: 192.168.200.130:9092	#主机IP:端口号
    producer:
      retries: 10
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

消费者

spring:
  kafka:
    bootstrap-servers: 192.168.200.130:9092	#主机IP:端口号
    consumer:
      group-id: ${spring.application.name}
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

3.生产者发送消息

@RestController
public class HelloController {
    @Autowired
    private KafkaTemplate<String,String> kafkaTemplate;
    //发送String
    @GetMapping("/hello")
    public String hello(){
        String message = "Hello,SpringBoot Kafka!";
        kafkaTemplate.send("topic",message);
        return "success";
    }
    //发送pojo
        @GetMapping("/sendPojo")
    public String sendPojo(){
        User user = User.builder()
                .id(1001)
                .name("张三")
                .birthday(new Date())
                .build();
        kafkaTemplate.send("pojo-topic", JSON.toJSONString(user));
        return "success";
    }
}

3.消费者接收消息

@Component
public class HelloListener {
    //接收Stirng
    @KafkaListener(topics = "topic")
    public void onMessage(String message){
        if(!StringUtils.isEmpty(message)){
            System.out.println(message);
        }
    }
    //接收pojo
        @KafkaListener(topics = "pojo-topic")
    public void onMessage2(String message) {
        if (!StringUtils.isEmpty(message)){
            User user = JSON.parseObject(message, User.class);
            System.out.println("user = " + user);
        }
    }
}

4.总结

#kafka安装(docker)

Kafka对于zookeeper是强依赖，保存kafka相关的节点数据，所以安装Kafka之前必须先安装zookeeper

标签 , ,