package com.niugang.config;

import java.util.HashMap;
import java.util.Map;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.consumer.OffsetCommitCallback;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.listener.AbstractMessageListenerContainer;

import com.niugang.controller.SenderConttoller;

/**
 * 
 * @ClassName: KafkaConfig
 * @Description:kafka配置類(lèi),基于spring java純配置的
 * @author: niugang
 * @date: 2018年10月20日 下午8:04:26
 * @Copyright: 863263957@qq.com. All rights reserved.
 *
 */
@Configuration
@EnableKafka
public class KafkaConfig {

  private Logger logger = LoggerFactory.getLogger(KafkaConfig.class);

  @Bean
  public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
    // 偏移量提交方式
    // factory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.COUNT);
    // 異步提交偏移量(默認(rèn)就是true)
    // factory.getContainerProperties().setSyncCommits(true);
    //回調(diào)函數(shù)經(jīng)常用于記錄提交錯(cuò)誤
    /*factory.getContainerProperties().setCommitCallback(new OffsetCommitCallback() {

      @Override
      public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
        if (exception != null) {
          logger.error("Commit failed for effsets {}", offsets, exception);
        }

      }
    });*/
    factory.setConsumerFactory(consumerFactory());
    return factory;
  }

  /**
   * 消費(fèi)者工廠配置
   * 
   * @return
   */
  @Bean
  public ConsumerFactory<String, String> consumerFactory() {
    return new DefaultKafkaConsumerFactory<>(consumerProps());
  }

  /**
   * 生產(chǎn)者工廠配置
   * 
   * @return
   */
  @Bean
  public ProducerFactory<String, String> producerFactory() {
    return new DefaultKafkaProducerFactory<>(senderProps());
  }

  /**
   * kafka發(fā)送消息模板
   * 
   * @return
   */
  @Bean
  public KafkaTemplate<String, String> kafkaTemplate() {
    return new KafkaTemplate<String, String>(producerFactory());
  }

  /**
   * 消費(fèi)者監(jiān)聽(tīng)
   * 
   * @return
   */
  @Bean
  public ConsumerListener listener() {
    return new ConsumerListener();
  }

  /**
   * 消費(fèi)配置方法
   * 
   * @return
   */
  private Map<String, Object> consumerProps() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "kafka_group_1");
    /**
     * enable.auto.commit 默認(rèn)5秒自動(dòng)提交偏移量
     */
    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
    props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100");
    props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
    /**
     * kafka是基于key-value鍵值對(duì)的，以下配置key和value的反序列化放
     */
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    return props;
  }

  /**
   * 生產(chǎn)者配置方法
   * 
   * 生產(chǎn)者有三個(gè)必選屬性
   * <p>
   * 1.bootstrap.servers broker地址清單，清單不要包含所有的broker地址，
   * 生產(chǎn)者會(huì)從給定的broker里查找到其他broker的信息。不過(guò)建議至少提供兩個(gè)broker信息，一旦 其中一個(gè)宕機(jī)，生產(chǎn)者仍能能夠連接到集群上。
   * </p>
   * <p>
   * 2.key.serializer broker希望接收到的消息的鍵和值都是字節(jié)數(shù)組。 生產(chǎn)者用對(duì)應(yīng)的類(lèi)把鍵對(duì)象序列化成字節(jié)數(shù)組。
   * </p>
   * <p>
   * 3.value.serializer 值得序列化方式
   * </p>
   * 
   * 
   * @return
   */
  private Map<String, Object> senderProps() {
    Map<String, Object> props = new HashMap<>();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    /**
     * 當(dāng)從broker接收到的是臨時(shí)可恢復(fù)的異常時(shí)，生產(chǎn)者會(huì)向broker重發(fā)消息，但是不能無(wú)限
     * 制重發(fā)，如果重發(fā)次數(shù)達(dá)到限制值，生產(chǎn)者將不會(huì)重試并返回錯(cuò)誤。
     * 通過(guò)retries屬性設(shè)置。默認(rèn)情況下生產(chǎn)者會(huì)在重試后等待100ms，可以通過(guò) retries.backoff.ms屬性進(jìn)行修改
     */
    props.put(ProducerConfig.RETRIES_CONFIG, 0);
    /**
     * 在考慮完成請(qǐng)求之前，生產(chǎn)者要求leader收到的確認(rèn)數(shù)量。這可以控制發(fā)送記錄的持久性。允許以下設(shè)置：
     * <ul>
     * <li>
     * <code> acks = 0 </ code>如果設(shè)置為零，則生產(chǎn)者將不會(huì)等待來(lái)自服務(wù)器的任何確認(rèn)。該記錄將立即添加到套接字緩沖區(qū)并視為已發(fā)送。在這種情況下，無(wú)法保證服務(wù)器已收到記錄，并且
     * <code>retries </ code>配置將不會(huì)生效（因?yàn)榭蛻?hù)端通常不會(huì)知道任何故障）。為每條記錄返回的偏移量始終設(shè)置為-1。
     * <li> <code> acks = 1 </code>
     * 這意味著leader會(huì)將記錄寫(xiě)入其本地日志，但無(wú)需等待所有follower的完全確認(rèn)即可做出回應(yīng)。在這種情況下，
     * 如果leader在確認(rèn)記錄后立即失敗但在關(guān)注者復(fù)制之前，則記錄將丟失。
     * <li><code> acks = all </code>
     * 這意味著leader將等待完整的同步副本集以確認(rèn)記錄。這保證了只要至少一個(gè)同步副本仍然存活，記錄就不會(huì)丟失。這是最強(qiáng)有力的保證。
     * 這相當(dāng)于acks = -1設(shè)置
     */
    props.put(ProducerConfig.ACKS_CONFIG, "1");
    /**
     * 當(dāng)有多條消息要被發(fā)送到統(tǒng)一分區(qū)是，生產(chǎn)者會(huì)把他們放到統(tǒng)一批里。kafka通過(guò)批次的概念來(lái) 提高吞吐量，但是也會(huì)在增加延遲。
     */
    // 以下配置當(dāng)緩存數(shù)量達(dá)到16kb，就會(huì)觸發(fā)網(wǎng)絡(luò)請(qǐng)求，發(fā)送消息
    props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
    // 每條消息在緩存中的最長(zhǎng)時(shí)間，如果超過(guò)這個(gè)時(shí)間就會(huì)忽略batch.size的限制，由客戶(hù)端立即將消息發(fā)送出去
    props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
    props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
    // key的序列化方式
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    // value序列化方式
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    return props;
  }
}

package com.niugang.config;

import org.springframework.kafka.annotation.KafkaListener;

/**
 * 
 * @ClassName: ConsumerListener  
 * @Description:消費(fèi)者監(jiān)聽(tīng) 
 * @author: niugang
 * @date:  2018年10月21日 下午2:05:21  
 * @Copyright: 863263957@qq.com. All rights reserved. 
 *
 */
public class ConsumerListener {
  /**
   * topicPattern:支持正則表達(dá)式
   * @param foo
   */
  @KafkaListener(id = "foo", topics = "annotated1")
  public void listen1(String foo) {
    System.out.println("接收消息為："+foo);
  }
}