Kafka學習--spring boot 整合kafka 指定分區發送和消費

一、啟動kafka

　　啟動kafka之前一定要啟動zookeeper，因為要使用kafka必須要使用zookeeper。

　　windows環境下啟動，直接使用kafka自帶的zookeeper：

　　E:\kafka_2.12-2.4.0\bin\windows  zookeeper-server-start.bat ..\..\config\zookeeper.properties

　　接下來啟動kafka

　　E:\kafka_2.12-2.4.0\bin\windows  kafka-server-start.bat ..\..\config\server.properties

二、spring boot整合kafka項目實例

1.導入的maven

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

 配置文件：

server.port=80

#kafka地址，可以有多個
spring.kafka.bootstrap-servers=127.0.0.1:9092

#------生產者配置文件---------
 #指定kafka消息體和key的編碼格式
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

#設置等待acks返回的機制，有三個值 # 0：不等待返回的acks（可能會丟數據，因為發送消息沒有了失敗重試機制，但是這是最低延遲） # 1：消息發送給kafka分區中的leader后就返回（如果follower沒有同步完成leader就宕機了，就會丟數據） # -1（默認）：等待所有follower同步完消息后再發送（絕對不會丟數據）
spring.kafka.producer.acks=-1


# 消息累計到batch-size的值后，才會發送消息，默認為16384
spring.kafka.producer.batch-size=16384

#如果kafka遲遲不發送消息（這里指的是消息沒堆積到指定數量），那么過了這個時間（單位：毫米）開始發送
spring.kafka.producer.properties.linger.ms=1

#設置緩沖區大小，默認是33554432 #這個緩沖區是kafka中兩個線程里的共享變量 #這個兩個線程是main和sender，main負責把消息發送到共享變量，sender從共享變量拉數據
spring.kafka.producer.buffer-memory=33554432


#失敗重試發送的次數
spring.kafka.producer.retries=2

#------消費者配置文件---------
 #指定kafka消息體和key的編碼格式
spring.kafka.consumer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.consumer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

#指定消費者組的group_id
spring.kafka.consumer.group-id=kafka_test

#kafka意外宕機時，再次開啟消息消費的策略，共有三種策略 #earliest：當各分區下有已提交的offset時，從提交的offset開始消費；無提交的offset時，從頭開始消費 #latest：當各分區下有已提交的offset時，從提交的offset開始消費；無提交的offset時，消費新產生的該分區下的數據、 #none：當所有分區都存在已提交的offset時，從offset后開始消費；只要有一個分區不存在已提交的offset，則拋出異常
spring.kafka.consumer.auto-offset-reset=earliest


#自動提交offset
spring.kafka.consumer.enable-auto-commit=true

#提交offset時間間隔
spring.kafka.consumer.auto-commit-interval=100

#消費監聽接口監聽的主題不存在時，默認會報錯因此要關掉這個
spring.kafka.listener.missing-topics-fatal=false

 2.創建topic

import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * 使用代碼創建的topic
 * 三個參數意思：topic的名稱；分區數量，新主題的復制因子；如果指定了副本分配，則為-1。
 */
@Configuration
public class KafkaTopic {

     @Bean
    public NewTopic batchTopic() {
        return new NewTopic("testTopic", 8, (short) 1);
    }
}

3.生產者代碼

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;

/**
 * kafka生產者代碼
 */
@RestController
public class ProductorController {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

/** * 指定在1分區發送內容 */
    @RequestMapping("/test")
    public String show() {
        kafkaTemplate.send("testTopic","1", "你好");
        return "發送成功";
    }

}

4.消費者代碼

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.KafkaListener;

import java.util.Optional;

/**
 * kafka消費者代碼
 */
@Configuration
public class KafkaConsumer {

 //配置topic和分區,可以配置多個
//topic為隊列名稱
//partitions表示值的的分區，這里指定了0和2分區
//partitionOffsets表示詳細的指定分區，partition表示那個分區，initialOffset表示Offset的初始位置
    @KafkaListener(topicPartitions =
        { @TopicPartition(topic = "testTopic",
                partitions = { "0", "2" },
                partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "4"))
 }
    )public void consumer(ConsumerRecord consumerRecord){
        Optional<Object> kafkaMassage = Optional.ofNullable(consumerRecord.value());
        if(kafkaMassage.isPresent()){
            Object o = kafkaMassage.get();
            System.out.println("接收到的消息是："+o);
        }

    }

}

測試結果：

 

"1"