kafka 核心概念及快速入门

痛定思痛。 2022-04-02 09:21 360阅读 0赞

Kafka

Kafka 核心概念

什么是 Kafka

Kafka是由Apache软件基金会开发的一个开源流处理平台，由Scala和Java编写。该项目的目标是为处理实时数据提供一个统一、高吞吐、低延迟的平台。其持久化层本质上是一个“按照分布式事务日志架构的大规模发布/订阅消息队列”，这使它作为企业级基础设施来处理流式数据非常有价值。此外，Kafka可以通过Kafka Connect连接到外部系统（用于数据输入/输出），并提供了Kafka Streams——一个Java流式处理库。该设计受事务日志的影响较大。

基本概念

Kafka是一个分布式数据流平台，可以运行在单台服务器上，也可以在多台服务器上部署形成集群。它提供了发布和订阅功能，使用者可以发送数据到Kafka中，也可以从Kafka中读取数据（以便进行后续的处理）。Kafka具有高吞吐、低延迟、高容错等特点。下面介绍一下Kafka中常用的基本概念：

  • Broker
    消息队列中常用的概念，在Kafka中指部署了Kafka实例的服务器节点。
  • Topic
    用来区分不同类型信息的主题。比如应用程序A订阅了主题t1，应用程序B订阅了主题t2而没有订阅t1，那么发送到主题t1中的数据将只能被应用程序A读到，而不会被应用程序B读到。
  • Partition
    每个topic可以有一个或多个partition（分区）。分区是在物理层面上的，不同的分区对应着不同的数据文件。Kafka使用分区支持物理上的并发写入和读取，从而大大提高了吞吐量。
  • Record
    实际写入Kafka中并可以被读取的消息记录。每个record包含了key、value和timestamp。
  • Producer
    生产者，用来向Kafka中发送数据（record）。
  • Consumer
    消费者，用来读取Kafka中的数据（record）。
  • Consumer Group
    一个消费者组可以包含一个或多个消费者。使用多分区+多消费者方式可以极大提高数据下游的处理速度。

kafka 核心名词解释

  • Topic(主题): 每一条发送到kafka集群的消息都可以有一个类别，这个类别叫做topic,不同的消息会进行分开存储，如果topic很大，可以分布到多个broker上，也可以这样理解：topic被认为是一个队列，每一条消息都必须指定它的topic，可以说我们需要明确把消息放入哪一个队列。对于传统的message queue而言，一般会删除已经被消费的消息，而Kafka集群会保留所有的消息，无论其被消费与否。当然，因为磁盘限制，不可能永久保留所有数据（实际上也没必要），因此Kafka提供两种策略删除旧数据。一是基于时间，二是基于Partition文件大小。
  • Broker(代理): 一台kafka服务器就可以称之为broker.一个集群由多个broker组成，一个broker可以有多个topic
  • Partition(分区): 为了使得kafka吞吐量线性提高，物理上把topic分成一个或者多个分区，每一个分区是一个有序的队列。且每一个分区在物理上都对应着一个文件夹，该文件夹下存储这个分区所有消息和索引文件。
    分区的表示: topic名字-分区的id每个日志文件都是一个Log Entry序列，每个Log Entry包含一个4字节整型数值（值为M+5），1个字节的”magic value”，4个字节的CRC校验码，然后跟M个字节的消息这个log entries并非由一个文件构成，而是分成多个segment，每个segment以该segment第一条消息的offset命名并以“.kafka”为后缀。另外会有一个索引文件，它标明了每个segment下包含的log entry的offset范围分区中每条消息都有一个当前Partition下唯一的64字节的offset，它指明了这条消息的起始位置，Kafka只保证一个分区的数据顺序发送给消费者，而不保证整个topic里多个分区之间的顺序
  • Replicas（副本）: 试想：一旦某一个Broker宕机，则其上所有的Partition数据都不可被消费，所以需要对分区备份。其中一个宕机后其它Replica必须要能继续服务并且即不能造成数据重复也不能造成数据丢失。
    如果没有一个Leader，所有Replica都可同时读/写数据，那就需要保证多个Replica之间互相（N×N条通路）同步数据，数据的一致性和有序性非常难保证，大大增加了Replication实现的复杂性，同时也增加了出现异常的几率。而引入Leader后，只有Leader负责数据读写，Follower只向Leader顺序Fetch数据（N条通路），系统更加简单且高效。
    每一个分区，根据复制因子N，会有N个副本，比如在broker1上有一个topic，分区为topic-1, 复制因子为2，那么在两个broker的数据目录里，就都有一个topic-1,其中一个是leader，一个replicas同一个Partition可能会有多个Replica，而这时需要在这些Replication之间选出一个Leader，Producer和Consumer只与这个Leader交互，其它Replica作为Follower从Leader中复制数据
  • Producer: Producer将消息发布到指定的topic中，同时，producer还需要指定该消息属于哪个partition
  • Consumer: 本质上kafka只支持topic，每一个consumer属于一个consumer group，每个consumer group可以包含多个consumer。发送到topic的消息只会被订阅该topic的每个group中的一个consumer消费。如果所有的consumer都具有相同的group，这种情况和queue很相似，消息将会在consumer之间均衡分配；如果所有的consumer都在不同的group中，这种情况就是广播模式，消息会被发送到所有订阅该topic的group中，那么所有的consumer都会消费到该消息。kafka的设计原理决定，对于同一个topic，同一个group中consumer的数量不能多于partition的数量，否则就会有consumer无法获取到消息。
  • Offset: Offset专指Partition以及User Group而言，记录某个user group在某个partiton中当前已经消费到达的位置。

kafka使用场景

目前主流使用场景基本如下：

  • 消息队列（MQ）
    在系统架构设计中，经常会使用消息队列（Message Queue）——MQ。MQ是一种跨进程的通信机制，用于上下游的消息传递，使用MQ可以使上下游解耦，消息发送上游只需要依赖MQ，逻辑上和物理上都不需要依赖其他下游服务。MQ的常见使用场景如流量削峰、数据驱动的任务依赖等等。在MQ领域，除了Kafka外还有传统的消息队列如ActiveMQ和RabbitMQ等。
  • 追踪网站活动
    Kafka最出就是被设计用来进行网站活动（比如PV、UV、搜索记录等）的追踪。可以将不同的活动放入不同的主题，供后续的实时计算、实时监控等程序使用，也可以将数据导入到数据仓库中进行后续的离线处理和生成报表等。
  • Metrics
    Kafka经常被用来传输监控数据。主要用来聚合分布式应用程序的统计数据，将数据集中后进行统一的分析和展示等。
  • 日志聚合
    很多人使用Kafka作为日志聚合的解决方案。日志聚合通常指将不同服务器上的日志收集起来并放入一个日志中心，比如一台文件服务器或者HDFS中的一个目录，供后续进行分析处理。相比于Flume和Scribe等日志聚合工具，Kafka具有更出色的性能。

kafka 集群搭建

安装kefka集群

由于kafka依赖zookeeper环境所以先安装zookeeper，zk安装

安装环境

  1. linux: CentSO-7.5_x64
  2. java: jdk1.8.0_191
  3. zookeeper: zookeeper3.4.10
  4. kafka: kafka_2.11-2.0.1
  6. # 下载
  7. $ wget http://mirrors.hust.edu.cn/apache/kafka/2.1.0/kafka_2.11-2.1.0.tgz
  9. # 解压
  10. $ tar -zxvf kafka_2.11-2.1.0.tgz
  12. # 编辑配置文件修改一下几个配置
  13. $ vim $KAFKA_HOME/config/server.properties
  15. # 每台服务器的broker.id都不能相同只能是数字
  16. broker.id=1
  18. # 修改为你的服务器的ip或主机名
  19. advertised.listeners=PLAINTEXT://node-1:9092
  21. # 设置zookeeper的连接端口,将下面的ip修改为你的IP称或主机名
  22. zookeeper.connect=node-1:2181,node-2:2181,node-3:2181

启动Kafka集群并测试

  1. $ cd $KAFKA_HOME
  3. # 分别在每个节点启动kafka服务(-daemon表示在后台运行)
  4. $ bin/kafka-server-start.sh -daemon config/server.properties
  6. # 创建一个名词为 test-topic 的 Topic，partitions 表示分区数量为3 --replication-factor 表示副本数量为2
  7. $ bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 2 --partitions 3 --topic test-topic
  9. # 查看topic
  10. $ bin/kafka-topics.sh --list --zookeeper localhost:2181
  12. # 查看topic状态
  13. $ bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test-topic
  15. # 查看topic详细信息
  16. $ bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test-topic
  18. # 修改topic信息
  19. $ bin/kafka-topics.sh --alter --topic test-topic --zookeeper localhost:2181 --partitions 5
  21. # 删除topic(简单的删除，只是标记删除）
  22. $ bin/kafka-topics.sh --delete --topic test-topic --zookeeper localhost:2181
  24. # 在一台服务器上创建一个 producer (生产者)
  25. $ bin/kafka-console-producer.sh --broker-list node-1:9092,node-2:9092,node-3:9092 --topic test-topic
  27. # 在一台服务器上创建一个 consumer (消费者)
  28. $ bin/kafka-console-consumer.sh --bootstrap-server node-2:9092,node-3:9092,node-4:9092 --topic test-topic --from-beginning
  30. # 现在可以在生产者的控制台输入任意字符就可以看到消费者端有消费消息。

java 客户端连接kafka

普通java形式

  • pom.xml




    org.apache.kafka
    kafka_2.11
    2.1.0



    junit
    junit
    4.12



    org.slf4j
    slf4j-api
    1.7.25



    org.slf4j
    slf4j-log4j12
    1.7.25


    org.apache.maven.plugins
    maven-compiler-plugin
    3.8.0

    1.8
    1.8
    UTF-8

  • JavaKafkaConsumer.java 消费者

    import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.apache.kafka.clients.consumer.ConsumerRecords;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
    import org.apache.kafka.clients.producer.Producer;
    import org.apache.kafka.common.TopicPartition;
    import org.apache.kafka.common.serialization.StringDeserializer;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;

    import java.util.Collection;
    import java.util.Collections;
    import java.util.Properties;

    /**


    • *

    • @author leone

    • @since 2018-12-26
      **/
      public class JavaKafkaConsumer {

      private static Logger logger = LoggerFactory.getLogger(JavaKafkaConsumer.class);

      private static Producer producer;

      private final static String TOPIC = “kafka-test-topic”;

      private static final String ZOOKEEPER_HOST = “node-2:2181,node-3:2181,node-4:2181”;

      private static final String KAFKA_BROKER = “node-2:9092,node-3:9092,node-4:9092”;

      private static Properties properties;

      static {

      1.  properties = new Properties();
      2.  properties.put("bootstrap.servers", KAFKA_BROKER);
      3.  properties.put("group.id", "test");
      4.  properties.put("enable.auto.commit", "true");
      5.  properties.put("auto.commit.interval.ms", "1000");
      6.  properties.put("key.deserializer", StringDeserializer.class.getName());
      7.  properties.put("value.deserializer", StringDeserializer.class.getName());

      }

      public static void main(String[] args) {

      1.  final KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
      3.  consumer.subscribe(Collections.singletonList(TOPIC), new ConsumerRebalanceListener() {
      5.      public void onPartitionsRevoked(Collection<TopicPartition> collection) {
      7.      }
      9.      public void onPartitionsAssigned(Collection<TopicPartition> collection) {
      10.          // 将偏移设置到最开始
      11.          consumer.seekToBeginning(collection);
      12.      }
      13.  });
      14.  while (true) {
      15.      ConsumerRecords<String, String> records = consumer.poll(100);
      16.      for (ConsumerRecord<String, String> record : records) {
      17.          logger.info("offset: {}, key: {}, value: {}", record.offset(), record.key(), record.value());
      18.      }
      19.  }

      }

    }

  • JavaKafkaProducer.java 生产者

    import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.Producer;
    import org.apache.kafka.clients.producer.ProducerRecord;
    import org.apache.kafka.common.serialization.StringSerializer;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;

    import java.util.Properties;
    import java.util.UUID;

    /**


    • *

    • @author leone

    • @since 2018-12-26
      **/
      public class JavaKafkaProducer {

      private static Logger logger = LoggerFactory.getLogger(JavaKafkaProducer.class);

      private static Producer producer;

      private final static String TOPIC = “kafka-test-topic”;

      private static final String ZOOKEEPER_HOST = “node-2:2181,node-3:2181,node-4:2181”;

      private static final String KAFKA_BROKER = “node-2:9092,node-3:9092,node-4:9092”;

      private static Properties properties;

      static {

      1.  properties = new Properties();
      2.  properties.put("bootstrap.servers", KAFKA_BROKER);
      3.  properties.put("acks", "all");
      4.  properties.put("retries", 0);
      5.  properties.put("batch.size", 16384);
      6.  properties.put("linger.ms", 1);
      7.  properties.put("buffer.memory", 33554432);
      8.  properties.put("key.serializer", StringSerializer.class.getName());
      9.  properties.put("value.serializer", StringSerializer.class.getName());

      }

      public static void main(String[] args) {

      1.  Producer<String, String> producer = new KafkaProducer<>(properties);
      3.  for (int i = 0; i < 200; i++) {
      4.      try {
      5.          Thread.sleep(1000);
      6.      } catch (InterruptedException e) {
      7.          e.printStackTrace();
      8.      }
      9.      String uuid = UUID.randomUUID().toString();
      10.      producer.send(new ProducerRecord<>(TOPIC, Integer.toString(i), uuid));
      11.      logger.info("send message success key: {}, value: {}", i, uuid);
      12.  }
      13.  producer.close();

      }

    }

  • KafkaClient.java

    import kafka.admin.AdminUtils;
    import kafka.admin.RackAwareMode;
    import kafka.server.ConfigType;
    import kafka.utils.ZkUtils;
    import org.apache.kafka.clients.admin.AdminClient;
    import org.apache.kafka.clients.admin.CreateTopicsResult;
    import org.apache.kafka.clients.admin.NewTopic;
    import org.apache.kafka.common.security.JaasUtils;
    import org.junit.Test;

    import java.util.*;

    /**


    • *

    • @author leone

    • @since 2018-12-26
      **/
      public class KafkaClient {

      private final static String TOPIC = “kafka-test-topic”;

      private static final String ZOOKEEPER_HOST = “node-2:2181,node-3:2181,node-4:2181”;

      private static final String KAFKA_BROKER = “node-2:9092,node-3:9092,node-4:9092”;

      private static Properties properties = new Properties();

      static {

      1.  properties.put("bootstrap.servers", KAFKA_BROKER);

      }

      /**

      • 创建topic
        */
        @Test
        public void createTopic() {
        AdminClient adminClient = AdminClient.create(properties);
        List newTopics = Arrays.asList(new NewTopic(TOPIC, 1, (short) 1));
        CreateTopicsResult result = adminClient.createTopics(newTopics);
        try {
        1.  result.all().get();
        } catch (Exception e) {
        1.  e.printStackTrace();
        }
        }
  1.     /**
  2.      * 创建topic
  3.      */
  4.     @Test
  5.     public void create() {
  6.         ZkUtils zkUtils = ZkUtils.apply(ZOOKEEPER_HOST, 30000, 30000, JaasUtils.isZkSecurityEnabled());
  7.         // 创建一个3个分区2个副本名为t1的topic
  8.         AdminUtils.createTopic(zkUtils, "t1", 3, 2, new Properties(), RackAwareMode.Enforced$.MODULE$);
  9.         zkUtils.close();
  10.     }
  12.     /**
  13.      * 查询topic
  14.      */
  15.     @Test
  16.     public void listTopic() {
  17.         ZkUtils zkUtils = ZkUtils.apply(ZOOKEEPER_HOST, 30000, 30000, JaasUtils.isZkSecurityEnabled());
  18.         // 获取 topic 所有属性
  19.         Properties props = AdminUtils.fetchEntityConfig(zkUtils, ConfigType.Topic(), "streaming-topic");
  21.         Iterator it = props.entrySet().iterator();
  22.         while (it.hasNext()) {
  23.             Map.Entry entry = (Map.Entry) it.next();
  24.             System.err.println(entry.getKey() + " = " + entry.getValue());
  25.         }
  26.         zkUtils.close();
  27.     }
  29.     /**
  30.      * 修改topic
  31.      */
  32.     @Test
  33.     public void updateTopic() {
  34.         ZkUtils zkUtils = ZkUtils.apply(ZOOKEEPER_HOST, 30000, 30000, JaasUtils.isZkSecurityEnabled());
  35.         Properties props = AdminUtils.fetchEntityConfig(zkUtils, ConfigType.Topic(), "log-test");
  36.         // 增加topic级别属性
  37.         props.put("min.cleanable.dirty.ratio", "0.4");
  38.         // 删除topic级别属性
  39.         props.remove("max.message.bytes");
  40.         // 修改topic 'test'的属性
  41.         AdminUtils.changeTopicConfig(zkUtils, "log-test", props);
  42.         zkUtils.close();
  44.     }
  46.     /**
  47.      * 删除topic 't1'
  48.      */
  49.     @Test
  50.     public void deleteTopic() {
  51.         ZkUtils zkUtils = ZkUtils.apply(ZOOKEEPER_HOST, 30000, 30000, JaasUtils.isZkSecurityEnabled());
  52.         AdminUtils.deleteTopic(zkUtils, "t1");
  53.         zkUtils.close();
  54.     }
  57. }

  • log4j.properties 日志配置

    log4j.rootLogger=info, stdout
    log4j.appender.stdout=org.apache.log4j.ConsoleAppender
    log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
    log4j.appender.stdout.layout.ConversionPattern=%5p [%t] - %m%n

基于spirngboot整合kafka

  • pom.xml

    <?xml version=”1.0” encoding=”UTF-8”?>



    spring-boot-kafka
    com.andy
    1.0.7.RELEASE

    jar
    4.0.0




    io.spring.platform
    platform-bom
    Cairo-SR5
    pom
    import







    org.springframework.boot
    spring-boot-starter-web



    org.springframework.boot
    spring-boot-starter-amqp



    org.springframework.kafka
    spring-kafka







    org.apache.maven.plugins
    maven-compiler-plugin
    3.7.0

    1.8
    1.8
    UTF-8




    org.springframework.boot
    spring-boot-maven-plugin
    2.0.3.RELEASE


    ZIP




    repackage

  • application.yml

    spring:
    application:

    1. name: spring-jms

    kafka:

    1. bootstrap-servers: node-2:9092,node-3:9092,node-4:9092
    2. producer:
    3.   retries:
    4.   batch-size: 16384
    5.   buffer-memory: 33554432
    6.   compressionType: snappy
    7.   acks: all
    8. consumer:
    9.   group-id: 0
    10.   auto-offset-reset: earliest
    11.   enable-auto-commit: true

  • Message.java 消息

    /**


    • *

    • @author leone

    • @since 2018-12-26
      **/
      @ToString
      public class Message {

      private Long id;

      private T message;

      private Date time;

      public Long getId() {

      1.  return id;

      }

      public void setId(Long id) {

      1.  this.id = id;

      }

      public T getMessage() {

      1.  return message;

      }

      public void setMessage(T message) {

      1.  this.message = message;

      }

      public Date getTime() {

      1.  return time;

      }

      public void setTime(Date time) {

      1.  this.time = time;

      }
      }

  • KafkaController.java 控制器

    import com.andy.jms.kafka.service.KafkaSender;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.PathVariable;
    import org.springframework.web.bind.annotation.RequestParam;
    import org.springframework.web.bind.annotation.RestController;

    /**


    • *

    • @author leone

    • @since 2018-12-26
      **/
      @Slf4j
      @RestController
      public class KafkaController {

      @Autowired
      private KafkaSender kafkaSender;

      @GetMapping(“/kafka/{topic}”)
      public String send(@PathVariable(“topic”) String topic, @RequestParam String message) {

      1.  kafkaSender.send(topic, message);
      2.  return "success";

      }

    }

  • KafkaReceiver.java

    import lombok.extern.slf4j.Slf4j;
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.springframework.kafka.annotation.KafkaListener;
    import org.springframework.stereotype.Component;

    import java.util.Optional;

    /**


    • *

    • @author leone
    • @since 2018-12-26
      **/
      @Slf4j
      @Component
      public class KafkaReceiver {
  1.     @KafkaListener(topics = {"order"})
  2.     public void listen(ConsumerRecord<?, ?> record) {
  3.         Optional<?> kafkaMessage = Optional.ofNullable(record.value());
  4.         if (kafkaMessage.isPresent()) {
  5.             Object message = kafkaMessage.get();
  6.             log.info("record:{}", record);
  7.             log.info("message:{}", message);
  8.         }
  9.     }
  10. }

  • KafkaSender.java

    import com.andy.jms.kafka.commen.Message;
    import com.fasterxml.jackson.core.JsonProcessingException;
    import com.fasterxml.jackson.databind.ObjectMapper;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.kafka.core.KafkaTemplate;
    import org.springframework.stereotype.Component;

    import java.util.Date;

    /**


    • *

    • @author leone

    • @since 2018-12-26
      **/
      @Slf4j
      @Component
      public class KafkaSender {

      @Autowired
      private KafkaTemplate kafkaTemplate;

      @Autowired
      private ObjectMapper objectMapper;

      /*

      • @param topic
      • @param body
        */
        public void send(String topic, Object body) {
        Message message = new Message<>();
        message.setId(System.currentTimeMillis());
        message.setMessage(body.toString());
        message.setTime(new Date());
        String content = null;
        try {
        1.  content = objectMapper.writeValueAsString(message);
        } catch (JsonProcessingException e) {
        1.  e.printStackTrace();
        }
        kafkaTemplate.send(topic, content);
        log.info(“send {} to {} success!”, message, topic);
        }
        }

  • 启动类

    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;

    /**

    • @author Leone
    • @since 2018-04-10
      **/
      @SpringBootApplication
      public class JmsApplication {
      public static void main(String[] args) {
      1.  SpringApplication.run(JmsApplication.class, args);
      }
      }

github地址

发表评论

表情：
评论列表 （有 0 条评论，360人围观）

还没有评论，来说两句吧...

相关阅读

    相关 Kafka：Kafka核心概念

    1 消息系统简介 1.1 为什么要用消息系统 ？ 解耦 各位系统之间通过消息系统这个统一的接口交换数据，无须了解彼此的存在； 冗余 部分消息系统具有消息持久化能

    女爷i女爷i/ 0 赞/ 60 阅读