常用腳本命名：

• topic 相關(guān)命令 ：
• 查詢topic列表 ：sh kafka-topics.sh --bootstrap-server localhost:9092 --list
• 創(chuàng)建topic （名稱：first 分區(qū)：1個 副本 3個）副本數(shù)量不能超過集群數(shù)量
  • sh kafka-topics.sh --bootstrap-server localhost:9092 --topic first --create --partitions 1 --replication-factor 3
• topic 信息
  • sh kafka-topics.sh --bootstrap-server localhost:9092 --topic first --describe
• 修改topic 分區(qū)數(shù)（只能增加）
  • sh kafka-topics.sh --bootstrap-server localhost:9092 --topic first --describe --partitions 3
• 生產(chǎn)消息：
  • sh kafka-console-producer.sh --bootstrap-server localhost:9092 --topic first
• 消費消費：
  • sh kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic first
  • sh kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic first --from-beginning

Spring boot 簡單整合：

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

server:
  port: 8200

spring:
  mvc:
    pathmatch:
      matching-strategy: ant_path_matcher
  application:
    name: @artifactId@
  kafka:
    bootstrap-servers:
      - 192.168.1.250:32010
    # 生產(chǎn)配置
    producer:
      #序列化方式
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      properties:
        linger.ms: 10 #sender 等待事件
         #ssl認(rèn)證配置相關(guān)
#        sasl.mechanism: PLAIN
#        security.protocol: SASL_PLAINTEXT
#        sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="admin";
      #緩存區(qū)大小 32m
      buffer-memory: 33554432
      #批次大小 16k
      batch-size: 16
      # ISR 全部應(yīng)答
      #acks: -1
      #事務(wù)ID前綴 ，配合 @Transactional ，保證多個消息的原子性
      #transaction-id-prefix: "transaction-id-xx"
    #消費配置
    consumer:
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      #group-id: xiaoshu-1
      enable-auto-commit: false
      # 從最早消息開始消費，但是消費后，會記錄offset、相同 group-id不會再次消費 
      # offset 是針對每個消費者組
      auto-offset-reset: earliest
      #批量消費,每次最多消費多少條
      #max-poll-records: 50
       #ssl認(rèn)證配置相關(guān)
#      properties:
#        sasl.mechanism: PLAIN
#        security.protocol: SASL_PLAINTEXT
#        sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="admin";

    listener:
      # 手動調(diào)用Acknowledgment.acknowledge()后立即提交
      ack-mode: manual
      #批量消費,配合 @KafkaListener - batch="true"
      #type: batch

生產(chǎn)：

	@Resource
    private KafkaTemplate<String,String> kafkaTemplate;
    //@Transactional(rollbackFor = RuntimeException.class)，配合 ack配置 實現(xiàn)多條消息發(fā)送，原子性
    @ApiOperation(value = "推送消息到kafak")
    @GetMapping("/sendMsg")
    public String sendMsg(String topic,String msg){
        kafkaTemplate.send(topic,msg).addCallback(success -> {
            if (success==null){
                System.out.println("消息發(fā)送失敗");
                return;
            }
            // 消息發(fā)送到的topic
            String topicName = success.getRecordMetadata().topic();
            // 消息發(fā)送到的分區(qū)
            int partition = success.getRecordMetadata().partition();
            // 消息在分區(qū)內(nèi)的offset
            long offset = success.getRecordMetadata().offset();
            System.out.println("發(fā)送消息成功:" + topic + "-" + partition + "-" + offset);
        }, failure -> {
            System.out.println("發(fā)送消息失敗:" + failure.getMessage());
        });
        return "ok";
    }

消費：文章來源地址http://www.zghlxwxcb.cn/news/detail-434853.html

@Configuration
public class KafkaConsumer {

    private static final String TOPIC_DLT=".DLT";

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    /**
     * 每個分區(qū)由消費者組種得一個消費者消費，每個消費者獨立
     * 分區(qū) -》 消費 、2分區(qū)2個消費監(jiān)聽
     * @param record
     * @param consumer
     */
    @KafkaListener(groupId = "group-1", topicPartitions ={ @TopicPartition(topic = "four",partitions = {"0"})},batch = "false")
    public void consumerTopic1(ConsumerRecord<String, String> record, Consumer consumer){
        String value = record.value();
        String topic1 = record.topic();
        long offset = record.offset();
        int partition = record.partition();
        try {
            log.info("收到消息："+value+"topic:"+topic1+"offset:"+offset+"分區(qū)"+partition);
            //TODO 異常，推送到 對應(yīng)死信 ↓
            //int i=1/0;
        } catch (Exception e) {
            System.out.println("commit failed");
            kafkaTemplate.send(topic1+TOPIC_DLT,value);
        } finally {
            consumer.commitAsync();
        }
    }

    @KafkaListener(groupId = "group-1", topicPartitions ={ @TopicPartition(topic = "four",partitions = {"1"})},batch = "false")
    public void consumerTopic2(ConsumerRecord<String, String> record, Consumer consumer){
        String value = record.value();
        String topic1 = record.topic();
        long offset = record.offset();
        int partition = record.partition();
        try {
            log.info("收到消息："+value+"topic:"+topic1+"offset:"+offset+"分區(qū)"+partition);
            //TODO 異常，推送到 對應(yīng)死信 ↓
            //int i=1/0;
        } catch (Exception e) {
            System.out.println("commit failed");
            kafkaTemplate.send(topic1+TOPIC_DLT,value);
        } finally {
            consumer.commitAsync();
        }
    }

}

	/**
     * 監(jiān)聽 topic1 ->轉(zhuǎn)發(fā)到 topic2
     */
    @KafkaListener(topics = {"topic1"},groupId = "group-4")
    @SendTo("topic2")
    public String onMessage7(ConsumerRecord<?, ?> record) {
        return record.value()+"-轉(zhuǎn)發(fā)消息";
    }

    @KafkaListener(topics = {"topic2"},groupId = "group-5")
    public void onMessage8(ConsumerRecord<?, ?> record) {
        System.out.println("收到轉(zhuǎn)發(fā)消息"+record.value());
    }

到了這里，關(guān)于Kafka 基礎(chǔ)整理、 Springboot 簡單整合的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！