Springboot-kafka的內建以及使用方法

1.添加依賴包

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

2.yml配置檔案配置

spring:
  application:
    name: kafka        #服務名稱
  kafka:
    bootstrap-servers: IP:9092  # kafka伺服器位址(可以多個,以"，"分割)
    consumer: # 指定一個預設的組名
      group-id: kafkaGroup
      # earliest:當各分區下有已送出的offset時，從送出的offset開始消費；無送出的offset時，從頭開始消費
      # latest:當各分區下有已送出的offset時，從送出的offset開始消費；無送出的offset時，消費新産生的該分區下的資料
      # none:topic各分區都存在已送出的offset時，從offset後開始消費；隻要有一個分區不存在已送出的offset，則抛出異常
      auto-offset-reset: earliest
      # key/value的反序列化
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    producer:
      # key/value的序列化
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      batch-size: 4096   # 批量抓取
      buffer-memory: 10240         # 緩存容量
      bootstrap-servers: IP:9092        # 伺服器位址，可以多個逗号分隔
           

3.使用demo

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.PartitionOffset;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.core.KafkaOperations;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author: houkai
 * @Date: 2019/11/26 14:01
 */
@RestController
@EnableKafka
public class KafkaController {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    private Logger log = LoggerFactory.getLogger(KafkaController.class);

    /**消息發送*/
    @GetMapping("kafka")
    public String testKafka() {
        int iMax = 10;
        for (int i = 1; i < iMax; i++) {  //循環發6次kafka消息
            //topic名稱  key   消息資料
            kafkaTemplate.send("test", "key_" + i, "data_" + i);
        }
        return "success";
    }

    /**
     * 訂閱者
     * 對于同一個topic，不同組中的訂閱者都會收到消息，即一個topic對應多個consumer,
     *      同一組中的訂閱者隻有一個consumer能夠收到消息，即一個topic對應一個consumer
     */
    @KafkaListener(id = "customerID1", topics = "test", groupId = "group1")
    public void receive(ConsumerRecord<?, ?> consumer) {
        log.info("groupID: 1  topic名稱:" + consumer.topic() + ",key:" + consumer.key() + ",分區位置:" + consumer.partition() + ", 下标" + consumer.offset() +"value:" + consumer.value());
    }
    /**訂閱者*/
    @KafkaListener(id = "customerID2", topics = "test", groupId = "group2")
    public void receive2(ConsumerRecord<?, ?> consumer) {
        log.info("groupID:2  topic名稱:" + consumer.topic() + ",key:" + consumer.key() + ",分區位置:" + consumer.partition() + ", 下标" + consumer.offset() +"value:" + consumer.value());
    }
    @KafkaListener(id = "customerID3", topics = "test", groupId = "group2")
    public void receive3(ConsumerRecord<?, ?> consumer) {
        log.info("groupID2  topic名稱:" + consumer.topic() + ",key:" + consumer.key() + ",分區位置:" + consumer.partition() + ", 下标" + consumer.offset() +"value:" + consumer.value());
    }

    /**使用execute ProducerRecord 發送消息 */
    @GetMapping("execute")
    public void execute(){
        kafkaTemplate.execute(new KafkaOperations.ProducerCallback(){

            @Override
            public Object doInKafka(Producer producer) {
                ProducerRecord producerRecord = new ProducerRecord<String, String>("test", "test_data");
                //send方法需要回調的寫法
                producer.send(producerRecord, new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                        //send success
                        if (null == e) {
                            log.info("send message success");
                            return;
                        }
                        log.error("send message failed");
                    }
                });
                return null;
            }
        });
    }
}