Kafka詳解

bin/kafka-topics.sh --create --zookeeper 192.168.65.60:2181 --replication-factor 1 --partitions 1 --topic test

bin/kafka-topics.sh --list --zookeeper 192.168.65.60:2181

bin/kafka-topics.sh --delete --topic test --zookeeper 192.168.65.60:2181

//指定往哪個broker（也就是伺服器）上發訊息
bin/kafka-console-producer.sh --broker-list 192.168.65.60:9092 --topic test 
>this is a msg
>this is a another msg 

bin/kafka-console-consumer.sh --bootstrap-server 192.168.65.60:9092 --topic test   

//這裡便凸顯了與傳統訊息中介軟體的不同，消費完，訊息依舊保留（預設保留在磁碟一週）
bin/kafka-console-consumer.sh --bootstrap-server 192.168.65.60:9092 --from-beginning --topic test

執行bin/kafka-console-consumer.sh 命令顯示所有的可選引數

bin/kafka-console-consumer.sh --bootstrap-server 192.168.65.60:9092 --whitelist "test|test-2"

bin/kafka-console-consumer.sh --bootstrap-server 192.168.65.60:9092  --consumer-property group.id=testGroup --topic test

bin/kafka-console-consumer.sh --bootstrap-server 192.168.65.60:9092 --consumer-property group.id=testGroup-2 --topic test 

bin/kafka-consumer-groups.sh --bootstrap-server 192.168.65.60:9092 --list 

bin/kafka-consumer-groups.sh --bootstrap-server 192.168.65.60:9092 --describe --group testGroup

//current-offset：當前消費組的已消費偏移量
//log-end-offset：主題對應分割區訊息的結束偏移量(HW)
//lag：當前消費組未消費的訊息數

bin/kafka-topics.sh --create --zookeeper 192.168.65.60:2181 --replication-factor 1 --partitions 2 --topic test1

bin/kafka-topics.sh --describe --zookeeper 192.168.65.60:2181 --topic test1

leader節點負責給定partition的所有讀寫請求。
replicas 表示某個partition在哪幾個broker上存在備份。不管這個幾點是不是」leader「，甚至這個節點掛了，也會列出。
isr 是replicas的一個子集，它只列出當前還存活著的，並且已同步備份了該partition的節點。

bin/kafka-topics.sh -alter --partitions 3 --zookeeper 192.168.65.60:2181 --topic test

queue模式：多個consumer從伺服器中讀取資料，訊息只會到達一個consumer。
publish-subscribe模式：訊息會被廣播給所有的consumer。

queue模式：所有的consumer都位於同一個consumer group 下。
publish-subscribe模式：所有的consumer都有著自己唯一的consumer group。

<dependency>
   <groupId>org.apache.kafka</groupId>
   <artifactId>kafka-clients</artifactId>
   <version>2.4.1</version>
</dependency>

public class MsgProducer {
    private final static String TOPIC_NAME = "my-replicated-topic";

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        /* *********************************設定部分******************************************************/
        Properties props = new Properties();
        //叢集架構存取叢集，防止單節點故障發不出去
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.65.60:9092,192.168.65.60:9093,192.168.65.60:9094");
         /*
         發出訊息持久化機制引數
        （1）acks=0： 表示producer不需要等待任何broker確認收到訊息的回覆，就可以繼續傳送下一條訊息。效能最高，但是最容易丟訊息。（海量資料紀錄檔的話推薦這個，丟些訊息其實並不影響）
        （2）acks=1： 至少要等待leader已經成功將資料寫入本地log，但是不需要等待所有follower是否成功寫入。就可以繼續傳送下一條訊息。這種情況下，如果follower沒有成功備份資料，而此時leader又掛掉，則訊息會丟失。
　　　　　　　　（一般場景都可以用在這個，適中，也是預設值）
        （3）acks=-1或all： 需要等待 min.insync.replicas(預設為1，推薦設定大於等於2，即多少個節點寫入成功即可) 這個引數設定的副本個數都成功寫入紀錄檔，這種策略
            會保證只要有一個備份存活就不會丟失資料。這是最強的資料保證。一般除非是金融級別，或跟錢打交道的場景才會使用這種設定。（金融場景，對訊息十分敏感，不允許丟訊息）
         */
        props.put(ProducerConfig.ACKS_CONFIG, "1");
         /*
        傳送失敗會重試，重試次數設定，預設重試間隔100ms，重試能保證訊息傳送的可靠性，但是也可能造成訊息重複傳送，比如網路抖動，所以需要在接收者那邊做好訊息接收的冪等性處理。
        */
        props.put(ProducerConfig.RETRIES_CONFIG, 3);
        //重試間隔設定
        props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 300);
　　　　 //因為kafka並不是採取，你寫一條，我就發一條的操作，而是在本機申請了，一塊緩衝記憶體。你寫入一條，優先是寫到了本地緩衝記憶體中
　　　　 //然後使用者端還會有一個執行緒不斷的從這個本地緩衝記憶體中拿資料放入，傳送緩衝區（batch緩衝區【16KB】），拉滿就會傳送。（這也是效能為什麼會高的原因之一，還會壓縮資料）
        //設定傳送訊息的本地緩衝區，如果設定了該緩衝區，訊息會先傳送到本地緩衝區，可以提高訊息傳送效能，預設值是33554432，即32MB
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        /*
        kafka本地執行緒會從緩衝區取資料，批次傳送到broker，
        設定批次傳送訊息的大小，預設值是16384，即16kb，就是說一個batch滿了16kb就傳送出去
        */
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        /*
        預設值是0，意思就是訊息必須立即被傳送，但這樣會影響效能
        一般設定10毫秒左右，就是說這個訊息傳送完後會進入原生的一個batch，如果10毫秒內，這個batch滿了16kb就會隨batch一起被傳送出去
        如果10毫秒內，batch沒滿，那麼也必須把訊息傳送出去，不能讓訊息的傳送延遲時間太長
        */
        props.put(ProducerConfig.LINGER_MS_CONFIG, 10);
        //把傳送的key從字串序列化為位元組陣列
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //把傳送訊息value從字串序列化為位元組陣列
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        Producer<String, String> producer = new KafkaProducer<String, String>(props);
        /* *********************************使用部分******************************************************/
        int msgNum = 5;
        final CountDownLatch countDownLatch = new CountDownLatch(msgNum);  //用於非同步的展示
        for (int i = 1; i <= msgNum; i++) {
            Order order = new Order(i, 100 + i, 1, 1000.00);
            /* *********************************指定分割區與不指定分割區*****************************************/
            //指定傳送分割區
            /*ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(TOPIC_NAME
                    , 0, order.getOrderId().toString(), JSON.toJSONString(order));*/
            //未指定傳送分割區，具體傳送的分割區計算公式：hash(key)%partitionNum
            ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(TOPIC_NAME
                    , order.getOrderId().toString(), JSON.toJSONString(order));

            /* *********************************同步傳送與非同步傳送（傳送完後面還有邏輯優先非同步）*****************************************/
            //等待訊息傳送成功的同步阻塞方法
            RecordMetadata metadata = producer.send(producerRecord).get();
            System.out.println("同步方式傳送訊息結果：" + "topic-" + metadata.topic() + "|partition-"
                    + metadata.partition() + "|offset-" + metadata.offset());

            //非同步回撥方式傳送訊息
            /*producer.send(producerRecord, new Callback() {
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (exception != null) {
                        System.err.println("傳送訊息失敗：" + exception.getStackTrace());

                    }
                    if (metadata != null) {
                        System.out.println("非同步方式傳送訊息結果：" + "topic-" + metadata.topic() + "|partition-"
                                + metadata.partition() + "|offset-" + metadata.offset());
                    }
                    countDownLatch.countDown();
                }
            });*/

            //業務...
        }
        countDownLatch.await(5, TimeUnit.SECONDS);
        producer.close();
    }
}

@Data
public class Order {
    private Integer orderId;
    private Integer productId;
    private Integer productNum;
    private Double orderAmount;

    public Order(Integer orderId, Integer productId, Integer productNum, Double orderAmount) {
        super();
        this.orderId = orderId;
        this.productId = productId;
        this.productNum = productNum;
        this.orderAmount = orderAmount;
    }

}

public class MsgConsumer {
    private final static String TOPIC_NAME = "my-replicated-topic";
    private final static String CONSUMER_GROUP_NAME = "testGroup";

    public static void main(String[] args) throws Exception {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.65.60:9092,192.168.65.60:9093,192.168.65.60:9094");
        // 消費分組名
        props.put(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP_NAME);
        // 是否自動提交offset，預設就是true
        /*props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        // 自動提交offset的間隔時間
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");*/
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        /*
        當消費主題的是一個新的消費組，或者指定offset的消費方式，offset不存在，那麼應該如何消費
        latest(預設) ：只消費自己啟動之後傳送到主題的訊息
        earliest：第一次從頭開始消費，以後按照消費offset記錄繼續消費，這個需要區別於consumer.seekToBeginning(每次都從頭開始消費)
        */
        //props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        /*
        consumer給broker傳送心跳的間隔時間，broker接收到心跳如果此時有rebalance發生會通過心跳響應將
        rebalance方案下發給consumer，這個時間可以稍微短一點，預設3s
        */
        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 1000);
        /*
        伺服器端broker多久感知不到一個consumer心跳就認為他故障了，會將其踢出消費組，對應的Partition也會被重新分配給其他consumer，預設是10秒
        */
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10 * 1000);


        //一次poll最大拉取訊息的條數，如果消費者處理速度很快，可以設定大點，如果處理速度一般，可以設定小點，預設500
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 50);
        /*
        如果兩次poll操作間隔超過了這個時間，broker就會認為這個consumer處理能力太弱，會將其踢出消費組，將分割區分配給別的consumer消費
　　　　　有種適者生存的感覺，把處理能力弱的(如2核4G比4核8G弱，而且沒在限定時間內來拿任務)驅逐。【要麼檢查處理過程是否能夠優化縮短時間，要麼調整一次性獲取的條數，要麼增大間隔時間（間隔時間預設的一般不建議調整）】
        */
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 30 * 1000);

        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);

        consumer.subscribe(Arrays.asList(TOPIC_NAME));
        // 消費指定分割區
        //consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));

        //訊息回溯消費
        /*consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));
        consumer.seekToBeginning(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));*/

        //指定offset消費
        /*consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));
　　　　　//指定從TOPIC_NAME的第10條開始消費
        consumer.seek(new TopicPartition(TOPIC_NAME, 0), 10);*/

        //從指定時間點開始消費
        /*List<PartitionInfo> topicPartitions = consumer.partitionsFor(TOPIC_NAME);
        //從1小時前開始消費
        long fetchDataTime = new Date().getTime() - 1000 * 60 * 60;
        Map<TopicPartition, Long> map = new HashMap<>();
　　　　　//將topic的所有partition拿出來
        for (PartitionInfo par : topicPartitions) {
            map.put(new TopicPartition(TOPIC_NAME, par.partition()), fetchDataTime);
        }
　　　　 //尋找每個partition的符合時間節點的offset
        Map<TopicPartition, OffsetAndTimestamp> parMap = consumer.offsetsForTimes(map);
　　　　 //針對每個partition進行消費
        for (Map.Entry<TopicPartition, OffsetAndTimestamp> entry : parMap.entrySet()) {
            TopicPartition key = entry.getKey();
            OffsetAndTimestamp value = entry.getValue();
            if (key == null || value == null) continue;
            Long offset = value.offset();
            System.out.println("partition-" + key.partition() + "|offset-" + offset);
            System.out.println();
            //根據消費裡的timestamp確定offset
            if (value != null) {
                consumer.assign(Arrays.asList(key));
                consumer.seek(key, offset);
            }
        }*/


        while (true) {
            /*
             * poll() API 是拉取訊息的長輪詢
             */
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("收到訊息：partition = %d,offset = %d, key = %s, value = %s%n", record.partition(),
                        record.offset(), record.key(), record.value());
            }

            if (records.count() > 0) {
                // 手動同步提交offset，當前執行緒會阻塞直到offset提交成功
                // 一般使用同步提交，因為提交之後一般也沒有什麼邏輯程式碼了
                //consumer.commitSync();

                // 手動非同步提交offset，當前執行緒提交offset不會阻塞，可以繼續處理後面的程式邏輯
                /*consumer.commitAsync(new OffsetCommitCallback() {
                    @Override
                    public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
                        if (exception != null) {
                            System.err.println("Commit failed for " + offsets);
                            System.err.println("Commit failed exception: " + exception.getStackTrace());
                        }
                    }
                });*/

            }
        }
    }
}

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

spring:
  kafka:
    bootstrap-servers: 192.168.65.60:9092,192.168.65.60:9093,192.168.65.60:9094
    producer: # 生產者
      retries: 3 # 設定大於0的值，則使用者端會將傳送失敗的記錄重新傳送
      batch-size: 16384
      buffer-memory: 33554432
      acks: 1
      # 指定訊息key和訊息體的編解碼方式
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer:
      group-id: default-group
      enable-auto-commit: false
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    listener:
      # 當每一條記錄被消費者監聽器（ListenerConsumer）處理之後提交
      # RECORD
      # 當每一批poll()的資料被消費者監聽器（ListenerConsumer）處理之後提交
      # BATCH
      # 當每一批poll()的資料被消費者監聽器（ListenerConsumer）處理之後，距離上次提交時間大於TIME時提交
      # TIME
      # 當每一批poll()的資料被消費者監聽器（ListenerConsumer）處理之後，被處理record數量大於等於COUNT時提交
      # COUNT
      # TIME |　COUNT　有一個條件滿足時提交
      # COUNT_TIME
      # 當每一批poll()的資料被消費者監聽器（ListenerConsumer）處理之後, 手動呼叫Acknowledgment.acknowledge()後提交
      # MANUAL
      # 手動呼叫Acknowledgment.acknowledge()後立即提交
      # MANUAL_IMMEDIATE
      ack-mode: MANUAL_IMMEDIATE

@RestController
public class KafkaController {

    private final static String TOPIC_NAME = "my-replicated-topic";

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @RequestMapping("/send")
    public void send() {
        kafkaTemplate.send(TOPIC_NAME, 0, "key", "this is a msg");
    }

}

@Component
public class MyConsumer {

    /**
     * @KafkaListener(groupId = "testGroup", topicPartitions = {
     *             @TopicPartition(topic = "topic1", partitions = {"0", "1"}),
     *             @TopicPartition(topic = "topic2", partitions = "0",
     *                     partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "100"))
     *     },concurrency = "6")
     *  //concurrency就是同組下的消費者個數，就是並行消費數，必須小於等於分割區總數
     * @param record
     */
    @KafkaListener(topics = "my-replicated-topic",groupId = "zGroup")
    public void listenZGroup(ConsumerRecord<String, String> record, Acknowledgment ack) {
        String value = record.value();
        System.out.println(value);
        System.out.println(record);
        //手動提交offset
        //ack.acknowledge();
    }

    //設定多個消費組
    /*@KafkaListener(topics = "my-replicated-topic",groupId = "tuGroup")
    public void listenTuGroup(ConsumerRecord<String, String> record, Acknowledgment ack) {
        String value = record.value();
        System.out.println(value);
        System.out.println(record);
        ack.acknowledge();
    }*/
}

當某個分割區的leader副本出現故障時，由控制器負責為該分割區選舉新的leader副本。
當檢測到某個分割區的ISR集合發生變化時，由控制器負責通知所有broker更新其後設資料資訊。
當使用kafka-topics.sh指令碼為某個topic增加分割區數量時，同樣還是由控制器負責讓新分割區被其他節點感知到。

1.監聽broker相關的變化。為Zookeeper中的/brokers/ids/節點新增BrokerChangeListener，用來處理broker增減的變化。
2.監聽topic相關的變化。為Zookeeper中的/brokers/topics節點新增TopicChangeListener，用來處理topic增減的變化；為Zookeeper中的/admin/delete_topics節點新增TopicDeletionListener，用來處理刪除topic的動作。
3.從Zookeeper中讀取獲取當前所有與topic、partition以及broker有關的資訊並進行相應的管理。對於所有topic所對應的Zookeeper中的/brokers/topics/[topic]節點新增PartitionModificationsListener，用來監聽topic中的分割區分配變化。
4.更新叢集的後設資料資訊，同步到其他普通的broker節點中。

1.副本節點不能產生分割區，必須能與zookeeper保持對談以及跟leader副本網路連通
2.副本能複製leader上的所有寫操作，並且不能落後太多。(與leader副本同步滯後的副本，是由 replica.lag.time.max.ms 設定決定的，超過這個時間都沒有跟leader同步過的一次的副本會被移出ISR列表)

公式：hash(consumerGroupId)  %  __consumer_offsets主題的分割區數

1.消費組裡的consumer增加或減少了
2.動態給topic增加了分割區
3.消費組訂閱了更多的topic

1）分割區的分配要儘可能均勻 。
2）分割區的分配儘可能與上次分配的保持相同。

consumer1除了原有的0~3，會再分配一個7
consumer2除了原有的4~6，會再分配8和9

1. 指定了 patition，則直接使用；
2. 未指定 patition 但指定 key，通過對 key 的 value 進行hash 選出一個 patition
3. patition 和 key 都未指定，使用輪詢選出一個 patition。

1. producer 先從 zookeeper 的 "/brokers/.../state" 節點找到該 partition 的 leader
2. producer 將訊息傳送給該 leader
3. leader 將訊息寫入本地 log
4. followers 從 leader pull 訊息，寫入本地 log 後 向leader 傳送 ACK
5. leader 收到所有 ISR 中的 replica 的 ACK 後，增加 HW（high watermark，最後 commit 的 offset） 並向 producer 傳送 ACK

AR（Assigned Repllicas）一個partition的所有副本（就是replica，不區分leader或follower）
ISR（In-Sync Replicas）能夠和 leader 保持同步的 follower + leader本身 組成的集合。
OSR（Out-Sync Relipcas）不能和 leader 保持同步的 follower 集合
公式：AR = ISR + OSR

同樣為了容忍 n 臺節點的故障，第一種方案需要 2n+1 個副本，而第二種方案只需要 n+1 個副本，而 Kafka 的每個分割區都有大量的資料，第一種方案會造成大量資料的冗餘。
雖然第二種方案的網路延遲會比較高，但網路延遲對 Kafka 的影響較小。

# 預設10000 即 10秒
replica.lag.time.max.ms
# 允許 follower 副本落後 leader 副本的訊息數量，超過這個數量後，follower 會被踢出 ISR
replica.lag.max.messages

# 部分訊息的offset索引檔案，kafka每次往分割區發4K(可設定)訊息就會記錄一條當前訊息的offset到index檔案，（有點類似於MYSQL的索引，B+樹的快速定位）
# 如果要定位訊息的offset會先在這個檔案裡快速定位，再去log檔案裡找具體訊息
00000000000000000000.index
# 訊息儲存檔案，主要存offset和訊息體
00000000000000000000.log
# 訊息的傳送時間索引檔案，kafka每次往分割區發4K(可設定)訊息就會記錄一條當前訊息的傳送時間戳與對應的offset到timeindex檔案，
# 如果需要按照時間來定位訊息的offset，會先在這個檔案裡查詢
00000000000000000000.timeindex

//另一份紀錄檔段檔案
00000000000005367851.index
00000000000005367851.log
00000000000005367851.timeindex
//又是一份紀錄檔段檔案
00000000000009936472.index
00000000000009936472.log
00000000000009936472.timeindex