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

特性

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

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

• docker安装zookeeper

  docker pull zookeeper:3.4.14
  

  创建容器

  docker run -d --name zookeeper -p 2181:2181 zookeeper:3.4.14
  

• Docker安装kafka

  下载镜像：

  docker pull wurstmeister/kafka:2.12-2.3.1
  

  创建容器

  docker run -d --name kafka 
  --env KAFKA_ADVERTISED_HOST_NAME=主机IP 
  --env KAFKA_ZOOKEEPER_CONNECT=主机IP:2181 
  --env KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://主机IP:9092 
  --env KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 
  --env KAFKA_HEAP_OPTS="-Xmx256M -Xms256M" 
  --restart always=true
  --net=host wurstmeister/kafka:2.12-2.3.1