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

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

    /**
     * 查詢(xún)所有主題
     *
     * @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("查詢(xún)主題錯(cuò)誤！");
        } 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保證消息在同一個(gè)分區(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,會(huì)等待 不推薦同步！
        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自定義主題，項(xiàng)目啟動(dòng)就創(chuàng)建，
    @Value("${spring.kafka.topic}")
    String topic;
    @Value("${spring.kafka.bootstrap-servers}")
    String[] server;
    /**
     * 項(xiàng)目啟動(dòng) 初始化主題,如果存在不會(huì)覆蓋主題的
     */
    @Bean
    public NewTopic batchTopic() {
        // 最大復(fù)制因子 <= 經(jīng)紀(jì)人broker數(shù)量.
        return new NewTopic(topic, 10, (short) server.length);
    }
}

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

    // 指定監(jiān)聽(tīng)分區(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");
        //確認(rèn)收到消息//確認(rèn)收到消息
        ack.acknowledge();
    }
    
    /**
     * @Title 指定topic、partition、offset消費(fèi)
     * @Description 同時(shí)監(jiān)聽(tīng)topic1和topic2，監(jiān)聽(tīng)topic1的0號(hào)分區(qū)、topic2的 "0號(hào)和1號(hào)" 分區(qū)，指向1號(hào)分區(qū)的offset初始值為8
     * 注意：topics和topicPartitions不能同時(shí)使用；
     **/
    @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) {
        //確認(rèn)收到消息
        ack.acknowledge();
    }
    
    /**        
     * 批量消費(fèi)監(jiān)聽(tīng)goods變更消息
     * yml配置listener:type 要改為batch
     * ymk配置consumer：max-poll-records: ？？（每次拉取多少條數(shù)據(jù)消費(fèi)）
     * concurrency = "2" 啟動(dòng)多少線(xiàn)程執(zhí)行，應(yīng)小于等于broker數(shù)量，避免資源浪費(fèi)
     */
    @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);
            }
        }
    }
}