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
  

標籤：Java

上一篇：《Terraform 101 從入門到實踐》 第一章 Terraform初相識

下一篇：【學習筆記】Http請求方法總結