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

kafka:
    #bootstrap-servers: server1:9092,server2:9093 #kafka開發(fā)地址,
    #生產(chǎn)者配置
    producer:
      # Kafka提供的序列化和反序列化類
      key-serializer: org.apache.kafka.common.serialization.StringSerializer #序列化
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      retries: 1 # 消息發(fā)送重試次數(shù)
      #acks = 0：設(shè)置成 表示 producer 完全不理睬 leader broker 端的處理結(jié)果。此時producer 發(fā)送消息后立即開啟下 條消息的發(fā)送，根本不等待 leader broker 端返回結(jié)果
      #acks= all 或者－1 ：表示當發(fā)送消息時， leader broker 不僅會將消息寫入本地日志，同時還會等待所有其他副本都成功寫入它們各自的本地日志后，才發(fā)送響應結(jié)果給，消息安全但是吞吐量會比較低。
      #acks = 1：默認的參數(shù)值。 producer 發(fā)送消息后 leader broker 僅將該消息寫入本地日志，然后便發(fā)送響應結(jié)果給producer ，而無須等待其他副本寫入該消息。折中方案，只要leader一直活著消息就不會丟失，同時也保證了吞吐量
      acks: 1 #應答級別:多少個分區(qū)副本備份完成時向生產(chǎn)者發(fā)送ack確認(可選0、1、all/-1)
      batch-size: 16384 #批量大小
      properties:
        linger:
          ms: 0 #提交延遲
      buffer-memory: 33554432 # 生產(chǎn)端緩沖區(qū)大小
    # 消費者配置
    consumer:
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      # 分組名稱
      group-id: web
      enable-auto-commit: false
      #提交offset延時(接收到消息后多久提交offset)
      # auto-commit-interval: 1000ms
      #當kafka中沒有初始offset或offset超出范圍時將自動重置offset
      # earliest:重置為分區(qū)中最小的offset;
      # latest:重置為分區(qū)中最新的offset(消費分區(qū)中新產(chǎn)生的數(shù)據(jù));
      # none:只要有一個分區(qū)不存在已提交的offset,就拋出異常;
      auto-offset-reset: latest
      properties:
        #消費會話超時時間(超過這個時間consumer沒有發(fā)送心跳,就會觸發(fā)rebalance操作)
        session.timeout.ms: 15000
        #消費請求超時時間
        request.timeout.ms: 18000
      #批量消費每次最多消費多少條消息
      #每次拉取一條，一條條消費，當然是具體業(yè)務狀況設(shè)置
      max-poll-records: 1
      # 指定心跳包發(fā)送頻率，即間隔多長時間發(fā)送一次心跳包，優(yōu)化該值的設(shè)置可以減少Rebalance操作，默認時間為3秒；
      heartbeat-interval: 6000
      # 發(fā)出請求時傳遞給服務器的 ID。用于服務器端日志記錄 正常使用后解開注釋，不然只有一個節(jié)點會報錯
      #client-id: mqtt
    listener:
      #消費端監(jiān)聽的topic不存在時，項目啟動會報錯(關(guān)掉)
      missing-topics-fatal: false
      #設(shè)置消費類型 批量消費 batch，單條消費：single
      type: single
      #指定容器的線程數(shù)，提高并發(fā)量
      #concurrency: 3
      #手動提交偏移量 manual達到一定數(shù)據(jù)后批量提交
      #ack-mode: manual
      ack-mode: MANUAL_IMMEDIATE #手動確認消息
        # 認證
    #properties:
      #security:
        #protocol: SASL_PLAINTEXT
      #sasl:
        #mechanism: SCRAM-SHA-256
        #jaas:config: 'org.apache.kafka.common.security.scram.ScramLoginModule required username="username" password="password";'

@Component
@Slf4j
public class KafkaUtils<K, V> {
    @Autowired
    private KafkaTemplate kafkaTemplate;
    @Value("${spring.kafka.bootstrap-servers}")
    String[] servers;

    /**
     * 獲取連接
     * @return
     */
    private Admin getAdmin() {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", servers);
        // 正式環(huán)境需要添加賬號密碼
        return Admin.create(properties);
    }

    /**
     * 增加topic
     *
     * @param name      主題名字
     * @param partition 分區(qū)數(shù)量
     * @param replica   副本數(shù)量
     * @date 2022-06-23 chens
     */
    public R addTopic(String name, Integer partition, Integer replica) {
        Admin admin = getAdmin();
        if (replica > servers.length) {
            return R.error("副本數(shù)量不允許超過Broker數(shù)量");
        }
        try {
            NewTopic topic = new NewTopic(name, partition, Short.parseShort(replica.toString()));
            admin.createTopics(Collections.singleton(topic));
        } finally {
            admin.close();
        }
        return R.ok();
    }

    /**
     * 刪除主題
     *
     * @param names 主題名字集合
     * @date 2022-06-23 chens
     */
    public void deleteTopic(List<String> names) {
        Admin admin = getAdmin();
        try {
            admin.deleteTopics(names);
        } finally {
            admin.close();
        }
    }

    /**
     * 查詢所有主題
     *
     * @date 2022-06-24 chens
     */
    public Set<String> queryTopic() {
        Admin admin = getAdmin();
        try {
            ListTopicsResult topics = admin.listTopics();
            Set<String> set = topics.names().get();
            return set;
        } catch (Exception e) {
            log.error("查詢主題錯誤！");
        } finally {
            admin.close();
        }
        return null;
    }

    // 向所有分區(qū)發(fā)送消息
    public ListenableFuture<SendResult<K, V>> send(String topic, @Nullable V data) {
        return kafkaTemplate.send(topic, data);
    }
    
    // 指定key發(fā)送消息，相同key保證消息在同一個分區(qū)
    public ListenableFuture<SendResult<K, V>> send(String topic, K key, @Nullable V data) {
        return kafkaTemplate.send(topic, key, data);
    }

    // 指定分區(qū)和key發(fā)送。
    public ListenableFuture<SendResult<K, V>> send(String topic, Integer partition, K key, @Nullable V data) {
        return kafkaTemplate.send(topic, partition, key, data);
    }
}

@GetMapping("/{topic}")
    public String test(@PathVariable String topic, @PathVariable Long index) throws ExecutionException, InterruptedException {

        ListenableFuture future = null;
        Chenshuang user = new Chenshuang(i, "陳爽", "123456", new Date());
        String s = JSON.toJSONString(user);
        KafkaUtils utils = new KafkaUtils();
        future = kafkaUtils.send(topic, s);
        // 異步回調(diào)，同步get,會等待 不推薦同步！
        future.addCallback(new ListenableFutureCallback() {
            @Override
            public void onFailure(Throwable ex) {
                System.out.println("發(fā)送失敗");
            }
            @Override
            public void onSuccess(Object result) {
                System.out.println("發(fā)送成功：" + result);
            }
        });
        return "發(fā)送成功";
    }

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

@Configuration
public class KafkaTopic {
    // yml自定義主題，項目啟動就創(chuàng)建，
    @Value("${spring.kafka.topic}")
    String topic;
    @Value("${spring.kafka.bootstrap-servers}")
    String[] server;
    /**
     * 項目啟動 初始化主題,如果存在不會覆蓋主題的
     */
    @Bean
    public NewTopic batchTopic() {
        // 最大復制因子 <= 經(jīng)紀人broker數(shù)量.
        return new NewTopic(topic, 10, (short) server.length);
    }
}

@Component
@Slf4j
public class Consumer {
    // 監(jiān)聽主題 分組a
    @KafkaListener(topics =("${spring.kafka.topic}") ,groupId = "a")
    public  void  getMessage(ConsumerRecord message, Acknowledgment ack) {
        //確認收到消息
        ack.acknowledge();
    }
    // 監(jiān)聽主題 分組a
    @KafkaListener(topics = ("${spring.kafka.topic}"),groupId = "a")
    public  void getMessage2(ConsumerRecord message, Acknowledgment ack) {
        //確認收到消息
        ack.acknowledge();
    }
    // 監(jiān)聽主題 分組b
    @KafkaListener(topics = ("${spring.kafka.topic}"),groupId = "b")
    public  void getMessage3(ConsumerRecord message, Acknowledgment ack) {
        //確認收到消息//確認收到消息
        ack.acknowledge();
    }
    // 監(jiān)聽主題 分組b
    @KafkaListener(topics = ("${spring.kafka.topic}"),groupId = "b")
    public  void getMessage4(ConsumerRecord message, Acknowledgment ack) {
        //確認收到消息//確認收到消息
        ack.acknowledge();
    }

    // 指定監(jiān)聽分區(qū)1的消息
    @KafkaListener(topicPartitions = {@TopicPartition(topic = ("${spring.kafka.topic}"),partitions = {"1"})})
    public void getMessage5(ConsumerRecord message, Acknowledgment ack) {
        Long id = JSONObject.parseObject(message.value().toString()).getLong("id");
        //確認收到消息//確認收到消息
        ack.acknowledge();
    }
    
    /**
     * @Title 指定topic、partition、offset消費
     * @Description 同時監(jiān)聽topic1和topic2，監(jiān)聽topic1的0號分區(qū)、topic2的 "0號和1號" 分區(qū)，指向1號分區(qū)的offset初始值為8
     * 注意：topics和topicPartitions不能同時使用；
     **/
    @KafkaListener(id = "c1",groupId = "c",topicPartitions = {
            @TopicPartition(topic = "t1", partitions = { "0" }),
            @TopicPartition(topic = "t2", partitions = "0", partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "8"))})
    public void getMessage6(ConsumerRecord record,Acknowledgment ack) {
        //確認收到消息
        ack.acknowledge();
    }
    
    /**        
     * 批量消費監(jiān)聽goods變更消息
     * yml配置listener:type 要改為batch
     * ymk配置consumer：max-poll-records: ？？（每次拉取多少條數(shù)據(jù)消費）
     * concurrency = "2" 啟動多少線程執(zhí)行，應小于等于broker數(shù)量，避免資源浪費
     */
    @KafkaListener(id="sync-modify-goods", topics = "${spring.kafka.topic}",concurrency = "4")
    public void getMessage7(List<ConsumerRecord<String, String>> records){
        for (ConsumerRecord<String, String> msg:records) {
            GoodsChangeMsg changeMsg = null;
            try {
                changeMsg = JSONObject.parseObject(msg.value(), GoodsChangeMsg.class);
                syncGoodsProcessor.handle(changeMsg);
            }catch (Exception exception) {
                log.error("解析失敗{}", msg, exception);
            }
        }
    }
}