基于kafka实现异步消息队列DEMO

墨蓝 2022-05-09 06:18 235阅读 0赞

Jar包导入

<dependency>
          <groupId>org.apache.kafka</groupId>
          <artifactId>kafka-clients</artifactId>
          <version>1.1.0</version>
        </dependency>
        
        <dependency>
          <groupId>org.springframework.kafka</groupId>
          <artifactId>spring-kafka</artifactId>
          <version>1.1.1.RELEASE</version>
        </dependency>
        
        <dependency>
          <groupId>org.apache.kafka</groupId>
          <artifactId>kafka_2.12</artifactId>
          <version>1.1.0</version>
        </dependency>

一，基于Java API实现异步消息发送

1，kafka topic操作

\* kafka topic控制台操作命令

//kafka后台启动命令
    sh ./bin/kafka-server-start.sh -daemon ../config/server.properties
    // kafka停止命令，
    sh ./bin/kafka-server-stop.sh
    // 创建topic命令
    sh ./bin/kafka-topics.sh --create --zookeeper 192.168.91.131:2181 --replication-factor 1 --partitions 2 --topic standloneTest
    // 查看topic列表
    sh ./bin/kafka-topics.sh --list --zookeeper 192.168.91.131:2181
    // 修改topic 
    sh ./bin/kafka-topics.sh --alert --zookeeper 192.168.91.131:2181 --partitions 1 --topic myTest
    // 查看topic信息
    sh ./bin/kafka-topics.sh --describe --zookeeper 192.168.91.131:2181 --topic myTest
    // 删除topic
    sh ./bin/kafka-topics.sh --delete --zookeeper 192.168.91.131:2181 --topic myTest
    // 控制台生产者
    sh ./bin/kafka-console-producer.sh --broker-list 192.168.91.131:9092 --topic myTest
    // 控制台消费者
    sh ./bin/kafka-console-consumer.sh --zookeeper 192.168.91.131:2181 --from-beginning --topic myTest

\* 基于Java API操作

import kafka.admin.AdminUtils;
    import kafka.admin.RackAwareMode;
    import kafka.server.ConfigType;
    import kafka.utils.ZkUtils;
    import org.apache.kafka.common.security.JaasUtils;
    import scala.collection.JavaConversions;
    
    import java.util.List;
    import java.util.Properties;
    
    public class KafkaTopicDemo {
    
        private static final String ZK_CONN = "192.168.91.128:2181,192.168.91.129:2181," +
                "192.168.91.130:2181";
    
        private static ZkUtils zkUtils = null;
    
        static {
            zkUtils = ZkUtils.apply(ZK_CONN, 30000, 30000,
                    JaasUtils.isZkSecurityEnabled());
        }
    
        // 创建topic
        public static void createTopic(String topicName, Integer partition, Integer replicationFactor) {
            // 判断当前topicName是否存在
            if (AdminUtils.topicExists(zkUtils, topicName)) {
                System.out.println("topic hase exists");
            } else {
                AdminUtils.createTopic(zkUtils, topicName, partition, replicationFactor,
                        new Properties(), RackAwareMode.Enforced$.MODULE$);
                zkUtils.close();
                System.out.println("create topic success");
            }
        }
    
        // 修改topic
        public static void updateTopic(String topicName) {
            if (AdminUtils.topicExists(zkUtils, topicName)) {
                // 查看topic配置信息
                Properties properties = AdminUtils.fetchEntityConfig(zkUtils, ConfigType.Topic(), topicName);
                System.out.println("oldParam : " + properties);
                properties.put("min.cleanable.dirty.ratio", "0.3");
                AdminUtils.changeTopicConfig(zkUtils, topicName, properties);
                Properties newProperties = AdminUtils.fetchEntityConfig(zkUtils, ConfigType.Topic(), topicName);
                System.out.println("newParam : " + newProperties);
                System.out.println("update success");
            } else {
                System.out.println("topic is not exists");
            }
        }
    
        // 删除topic
        public static void deleteTopic(String topicName) {
            if (AdminUtils.topicExists(zkUtils, topicName)) {
                AdminUtils.deleteTopic(zkUtils, topicName);
                System.out.println("topic delete success");
            } else {
                System.out.println("topic is not exists");
            }
        }
    
        // 查看topic列表
        public static void findAllTopics() {
            List<String> lstTopicName = JavaConversions.seqAsJavaList(zkUtils.getAllTopics());
            System.out.println(lstTopicName);
        }
    
    }

2，topic级别配置信息解读

<table> 
 <thead> 
  <tr> 
   <th style="width:155px;">Property(属性)</th> 
   <th style="width:126px;">Default(默认值)</th> 
   <th style="width:205px;">Server Property(配置属性)</th> 
   <th style="width:365px;">Description(说明)</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td style="vertical-align:top;"> <p>cleanup.policy</p> </td> 
   <td style="vertical-align:top;"> <p>delete</p> </td> 
   <td style="vertical-align:top;"> <p>log.cleanup.policy</p> </td> 
   <td style="vertical-align:top;"> <p>日志清理策略选择有：delete和compact主要针对过期数据的处理，或是日志文件达到限制的额度，会被&nbsp; topic创建时的指定参数覆盖</p> </td> 
  </tr> 
  <tr> 
   <td style="width:155px;">delete.retention.ms</td> 
   <td style="width:126px;">86400000 (24 hours)</td> 
   <td style="width:205px;">log.cleaner.delete.retention.ms</td> 
   <td style="width:365px;">对于压缩的日志保留的最长时间，也是客户端消费消息的最长时间，同log.retention.minutes的区别在于一个控制未压缩数据，一个控制压缩后的数据。会被topic创建时的指定参数覆盖</td> 
  </tr> 
  <tr> 
   <td style="width:155px;">flush.messages</td> 
   <td style="width:126px;">None</td> 
   <td style="width:205px;">log.flush.interval.messages</td> 
   <td style="width:365px;">log文件”sync”到磁盘之前累积的消息条数,因为磁盘IO操作是一个慢操作,但又是一个”数据可靠性"的必要手段,所以此参数的设置,需要在"数据可靠性"与"性能"之间做必要的权衡.如果此值过大,将会导致每次"fsync"的时间较长(IO阻塞),如果此值过小,将会导致"fsync"的次数较多,这也意味着整体的client请求有一定的延迟.物理server故障,将会导致没有fsync的消息丢失.</td> 
  </tr> 
  <tr> 
   <td style="width:155px;">flush.ms</td> 
   <td style="width:126px;">None</td> 
   <td style="width:205px;">log.flush.interval.ms</td> 
   <td style="width:365px;">仅仅通过interval来控制消息的磁盘写入时机,是不足的.此参数用于控制"fsync"的时间间隔,如果消息量始终没有达到阀值,但是离上一次磁盘同步的时间间隔达到阀值,也将触发.</td> 
  </tr> 
  <tr> 
   <td style="width:155px;">index.interval.bytes</td> 
   <td style="width:126px;">4096</td> 
   <td style="width:205px;">log.index.interval.bytes</td> 
   <td style="width:365px;">当执行一个fetch操作后，需要一定的空间来扫描最近的offset大小，设置越大，代表扫描速度越快，但是也更好内存，一般情况下不需要搭理这个参数</td> 
  </tr> 
  <tr> 
   <td style="width:155px;">message.max.bytes</td> 
   <td style="width:126px;">1,000,000</td> 
   <td style="width:205px;">message.max.bytes</td> 
   <td style="width:365px;">表示消息的最大大小，单位是字节</td> 
  </tr> 
  <tr> 
   <td style="width:155px;">min.cleanable.dirty.ratio</td> 
   <td style="width:126px;">0.5</td> 
   <td style="width:205px;">log.cleaner.min.cleanable.ratio</td> 
   <td style="width:365px;">日志清理的频率控制，越大意味着更高效的清理，同时会存在一些空间上的浪费，会被topic创建时的指定参数覆盖</td> 
  </tr> 
  <tr> 
   <td style="width:155px;">retention.bytes</td> 
   <td style="width:126px;">None</td> 
   <td style="width:205px;">log.retention.bytes</td> 
   <td style="width:365px;">topic每个分区的最大文件大小，一个topic的大小限制&nbsp; =&nbsp;分区数*log.retention.bytes。-1没有大小限log.retention.bytes和log.retention.minutes任意一个达到要求，都会执行删除，会被topic创建时的指定参数覆盖</td> 
  </tr> 
  <tr> 
   <td style="width:155px;">retention.ms</td> 
   <td style="width:126px;">None</td> 
   <td style="width:205px;">log.retention.minutes</td> 
   <td style="width:365px;"> <p>数据存储的最大时间超过这个时间会根据log.cleanup.policy设置的策略处理数据，也就是消费端能够多久去消费数据，log.retention.bytes和log.retention.minutes达到要求，都会执行删除，会被topic创建时的指定参数覆盖</p> </td> 
  </tr> 
  <tr> 
   <td style="width:155px;">segment.bytes</td> 
   <td style="width:126px;">1 GB</td> 
   <td style="width:205px;">log.segment.bytes</td> 
   <td style="width:365px;">topic的分区是以一堆segment文件存储的，这个控制每个segment的大小，会被topic创建时的指定参数覆盖</td> 
  </tr> 
  <tr> 
   <td style="width:155px;">segment.index.bytes</td> 
   <td style="width:126px;">10 MB</td> 
   <td style="width:205px;">log.index.size.max.bytes</td> 
   <td style="width:365px;">对于segment日志的索引文件大小限制，会被topic创建时的指定参数覆盖</td> 
  </tr> 
  <tr> 
   <td style="width:155px;">log.roll.hours</td> 
   <td style="width:126px;">7 days</td> 
   <td style="width:205px;">log.roll.hours</td> 
   <td style="width:365px;">这个参数会在日志segment没有达到log.segment.bytes设置的大小，也会强制新建一个segment会被&nbsp; topic创建时的指定参数覆盖</td> 
  </tr> 
 </tbody> 
</table>

3，Producer发送消息

import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.ProducerConfig;
    import org.apache.kafka.clients.producer.ProducerRecord;
    
    import java.util.Properties;
    
    public class KafkaProducerDemo extends Thread {
    
        private final KafkaProducer<Integer, String> producer;
    
        private final String topic;
    
        public KafkaProducerDemo(String topic) {
            Properties properties = new Properties();
            properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
                    "192.168.91.129:9092,192.168.91.129:9092,192.168.91.130:9092");
            properties.put(ProducerConfig.CLIENT_ID_CONFIG, "KafkaProducerDemo");
            properties.put(ProducerConfig.ACKS_CONFIG, "-1");
            properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                    "org.apache.kafka.common.serialization.IntegerSerializer");
            properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                    "org.apache.kafka.common.serialization.StringSerializer");
            producer = new KafkaProducer<Integer, String>(properties);
            this.topic = topic;
        }
    
        @Override
        public void run() {
            for (int i = 0; i < 10; i++) {
                String message = "producer message : " + i;
                System.out.println(message);
                producer.send(new ProducerRecord<Integer, String>(topic, message));
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    
        public static void main(String[] args) {
            KafkaProducerDemo producerDemo = new KafkaProducerDemo("clusterTest");
            producerDemo.start();
        }
    
    }

4，Producer端配置信息解读

<table> 
 <thead> 
  <tr> 
   <th style="width:187px;">Property(属性)</th> 
   <th style="width:692px;">Description(说明)</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td style="width:187px;">bootstrap.servers</td> 
   <td style="width:692px;">broker的地址</td> 
  </tr> 
  <tr> 
   <td style="width:187px;">key.serializer</td> 
   <td style="width:692px;">关键字的序列化方式</td> 
  </tr> 
  <tr> 
   <td style="width:187px;">value.serializer</td> 
   <td style="width:692px;">消息值的序列化方式</td> 
  </tr> 
  <tr> 
   <td style="width:187px;">acks</td> 
   <td style="width:692px;">服务端向生产者发送响应方式，是否基于副本间信息同步后响应</td> 
  </tr> 
  <tr> 
   <td style="width:187px;">buffer.memory</td> 
   <td style="width:692px;">生产者的内存缓冲区大小。如果生产者发送消息的速度 &gt; 消息发送到kafka的速度，那么消息就会在缓冲区堆积，导致缓冲区不足。这个时候，send()方法要么阻塞，要么抛出异常。</td> 
  </tr> 
  <tr> 
   <td style="width:187px;">max.block.ms</td> 
   <td style="width:692px;">表示send()方法在抛出异常之前可以阻塞多久的时间，默认是60s</td> 
  </tr> 
  <tr> 
   <td style="width:187px;">compression.type</td> 
   <td style="width:692px;">消息在发往kafka之前可以进行压缩处理，以此来降低存储开销和网络带宽。默认值是null，可选的压缩算法有snappy、gzip和lz4</td> 
  </tr> 
  <tr> 
   <td style="width:187px;">retries</td> 
   <td style="width:692px;">生产者向kafka发送消息可能会发生错误，有的是临时性的错误，比如网络突然阻塞了一会儿，有的不是临时的错误，比如“消息太大了”，对于出现的临时错误，可以通过重试机制来重新发送</td> 
  </tr> 
  <tr> 
   <td style="width:187px;">retry.backoff.ms</td> 
   <td style="width:692px;">每次重试之间间隔的时间，第一次失败了，那么休息一会再重试，休息多久，可以通过这个参数来调节</td> 
  </tr> 
  <tr> 
   <td style="width:187px;">batch.size</td> 
   <td style="width:692px;">生产者在发送消息时，可以将即将发往同一个分区的消息放在一个批次里，然后将这个批次整体进行发送，这样可以节约网络带宽，提升性能。该参数就是用来规范一个批次的大小的。但是生产者并不是说要等到一个批次装满之后，才会发送，不是这样的，有时候半满，甚至只有一个消息的时候，也可能会发送，具体怎么选择，我们不知道，但是不是说非要等装满才发。因此，如果把该参数调的比较大的话，是不会造成消息发送延迟的，但是会占用比较大的内存。但是如果设置的太小，会造成消息发送次数增加，会有额外的IO开销</td> 
  </tr> 
  <tr> 
   <td style="width:187px;">linger.ms</td> 
   <td style="width:692px;">生产者在发送一个批次之前，可以适当的等一小会，这样可以让更多的消息加入到该批次。这样会造成延时增加，但是降低了IO开销，增加了吞吐量</td> 
  </tr> 
  <tr> 
   <td style="width:187px;">client.id</td> 
   <td style="width:692px;">服务器用来标志消息的来源，是一个任意的字符串</td> 
  </tr> 
  <tr> 
   <td style="width:187px;">max.in.flight.requests.per.connection</td> 
   <td style="width:692px;">一个消息发送给kafka集群，在收到服务端的响应之前的这段时间里，生产者还可以发n-1个消息</td> 
  </tr> 
  <tr> 
   <td style="width:187px;">request.timeout.ms</td> 
   <td style="width:692px;">生产者在发送消息之后，到收到服务端响应时，等待的时间限制</td> 
  </tr> 
  <tr> 
   <td style="width:187px;">max.request.size</td> 
   <td style="width:692px;">生产者发送消息的大小。可以是一个消息的大小，也可以发送的一个批次的消息大小</td> 
  </tr> 
  <tr> 
   <td style="width:187px;">eceive.buffer.bytes和send.buffer.bytes</td> 
   <td style="width:692px;">tcp socket接收和发送消息的缓冲区大小，其实指的就是ByteBuffer的大小</td> 
  </tr> 
 </tbody> 
</table>

5，Consumer消费消息

import org.apache.kafka.clients.consumer.ConsumerConfig;
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.apache.kafka.clients.consumer.ConsumerRecords;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
    
    import java.util.Collections;
    import java.util.Properties;
    
    public class KafkaConsumerDemo extends Thread {
    
        private final KafkaConsumer<String, Integer> kafkaConsumer;
    
        public KafkaConsumerDemo(String topic) {
            Properties properties = new Properties();
            properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                    "192.168.91.129:9092,192.168.91.129:9092,192.168.91.130:9092");
            properties.put(ConsumerConfig.GROUP_ID_CONFIG, "KafkaConsumerDemo");
            properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
            properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
            properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                    "org.apache.kafka.common.serialization.IntegerDeserializer");
            properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                    "org.apache.kafka.common.serialization.StringDeserializer");
            // 从第一个顺序读取
            properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
            kafkaConsumer = new KafkaConsumer<String, Integer>(properties);
            kafkaConsumer.subscribe(Collections.singleton(topic));
        }
    
        @Override
        public void run() {
            while (true) {
                ConsumerRecords<String, Integer> consumerRecords = kafkaConsumer.poll(1000);
                for (ConsumerRecord record : consumerRecords) {
                    System.out.println("consumer receive : " + record.value());
                }
            }
        }
    
        public static void main(String[] args) {
            KafkaConsumerDemo consumerDemo = new KafkaConsumerDemo("clusterTest");
            consumerDemo.start();
        }
    }

6，Consumer端配置信息解读

<table> 
 <thead> 
  <tr> 
   <th style="width:225px;">Property(属性)</th> 
   <th>Description(说明)</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td style="width:225px;">fetch.min.bytes</td> 
   <td>消费者从服务器获取记录的最小字节数，broker收到消费者拉取数据的请求的时候，如果可用数据量小于设置的值，那么broker将会等待有足够可用的数据的时候才返回给消费者，这样可以降低消费者和broker的工作负载，因为当主题不是很活跃的情况下，就不需要来来回回的处理消息，如果没有很多可用数据，但消费者的CPU 使用率却很高，那么就需要把该属性的值设得比默认值大。如果消费者的数量比较多，把该属性的值设置得大一点可以降低broker 的工作负载。</td> 
  </tr> 
  <tr> 
   <td style="width:225px;">fetch.max.wait.ms</td> 
   <td>fetch.min.bytes设置了broker返回给消费者最小的数据量，而fetch.max.wait.ms设置的则是broker的等待时间，两个属性只要满足了任何一条，broker都会将数据返回给消费者，也就是说举个例子，fetch.min.bytes设置成1MB，fetch.max.wait.ms设置成1000ms，那么如果在1000ms时间内，如果数据量达到了1MB，broker将会把数据返回给消费者；如果已经过了1000ms，但是数据量还没有达到1MB，那么broker仍然会把当前积累的所有数据返回给消费者。</td> 
  </tr> 
  <tr> 
   <td style="width:225px;">max.partition.fetch.bytes</td> 
   <td>该属性指定了服务器从每个分区里返回给消费者的最大字节数。它的默认值是lMB ， 也<br> 就是说，kafkaConsumer.poll() 方法从每个分区里返回的记录最多不超max.partitions.fetch.bytes&nbsp;指定的字节。如果一个主题有20 个分区和5 个消费者，那么每个消费者需要至少4MB 的可用内存来接收记录。在为消费者分配内存时，可以给它们多分配一些，因为如果群组里有消费者发生崩愤，剩下的消费者需要处理更多的分区。max.partition.fetch.bytes 的值必须比broker 能够接收的最大消息的字节数（通过max.message.size 属性配置）大， 否则消费者可能无法读取这些消息，导致消费者一直挂起重试，例如，max.message.size设置为2MB，而该属性设置为1MB，那么当一个生产者可能就会生产一条大小为2MB的消息，那么就会出现问题，消费者能从分区取回的最大消息大小就只有1MB，但是数据量是2MB，所以就会导致消费者一直挂起重试。在设置该属性时，另一个需要考虑的因素是消费者处理数据的时间。消费者需要频繁调用poll（）方法<br> 来避免会话过期和发生分区再均衡，如果单次调用poll （） 返回的数据太多，消费者需要更多的时间来处理，可能无怯及时进行下一个轮询来避免会话过期。如果出现这种情况， 可以把max.partitioin.fetch.bytes 值改<strong>小</strong>，或者延长会话过期时间。</td> 
  </tr> 
  <tr> 
   <td style="width:225px;">session.timeout.ms</td> 
   <td>该属性指定了当消费者被认为已经挂掉之前可以与服务器断开连接的时间。默认是3s，消费者在3s之内没有再次向服务器发送心跳，那么将会被认为已经死亡。此时，协调器将会出发再均衡，把它的分区分配给其他的消费者，该属性与heartbeat.interval.ms紧密相关，该参数定义了消费者发送心跳的时间间隔，也就是心跳频率，一般要同时修改这两个参数，heartbeat.interval.ms参数值必须要小于session.timeout.ms，一般是session.timeout.ms的三分之一，比如，session.timeout.ms设置成3min，那么heartbeat.interval.ms一般设置成1min，这样，可以更快的检测以及恢复崩溃的节点，不过长时间的轮询或垃圾收集可能导致非预期的再均衡（<strong>有一种情况就是网络延迟，本身消费者是没有挂掉的，但是网络延迟造成了心跳超时，这样本不该发生再均衡，但是因为网络原因造成了非预期的再均衡</strong>），把该属性的值设置得大一些，可以减少意外的再均衡，不过检测节点崩愤－需要更长的时间。</td> 
  </tr> 
  <tr> 
   <td style="width:225px;">auto.offset.reset</td> 
   <td>该属性指定了消费者在读取一个没有偏移量后者偏移量无效（<u>消费者长时间失效当前的偏移量已经过时并且被删除了</u>）的分区的情况下，应该作何处理，默认值是latest，也就是从最新记录读取数据（<strong>消费者启动之后生成的记录</strong>），另一个值是earliest，意思是在偏移量无效的情况下，消费者从起始位置开始读取数据。</td> 
  </tr> 
  <tr> 
   <td style="width:225px;">enable.auto.commit</td> 
   <td>指定了消费者是否自动提交偏移量，默认值是true，为了尽量避免重复数据和数据丢失，可以把它设置为false，有自己控制合适提交偏移量，<strong>如果设置为true， 可以通过设置&nbsp;auto.commit.interval.ms属性来控制提交的频率</strong></td> 
  </tr> 
  <tr> 
   <td style="width:225px;">partition.assignment.strategy</td> 
   <td> <p>分区分配策略，kafka又两个默认策略，</p> 
    <ol> 
     <li>Range：该策略会把主题的<strong>若干个连续</strong>的分区分配给消费者</li> 
     <li>Robin：该策略把主题的所有分区<strong>逐个</strong>分配给消费者</li> 
    </ol><p>分区策略默认是：org.apache.kafka.clients.consumer.RangeAssignor=&gt;Range策略</p> <p>org.apache.kafka.clients.consumer.RoundRobinAssignor=&gt;Robin策略</p> </td> 
  </tr> 
  <tr> 
   <td style="width:225px;">client.id</td> 
   <td>表示从客户端发来的消息</td> 
  </tr> 
  <tr> 
   <td style="width:225px;">max.poll.records</td> 
   <td>控制单次调用call方法能够返回的记录数量，帮助控制在轮询里需要处理的数据量。</td> 
  </tr> 
  <tr> 
   <td style="width:225px;"> <p>receive.buffer.bytes</p> <p>+</p> <p>send.buffer.bytes</p> </td> 
   <td>socket 在读写数据时用到的<strong>TCP 缓冲区</strong>也可以设置大小。如果它们被设为<strong>-1</strong>&nbsp;，就使用操作系统的默认值。如果生产者或消费者与broker 处于不同的数据中心内，可以适当增大这些值，因为跨数据中心的网络一般都有比较高的延迟和比较低的带宽</td> 
  </tr> 
 </tbody> 
</table>

7，配置信息链接

\* 详细配置信息，参考官网 [http://kafka.apache.org/documentation/\#producerapi][http_kafka.apache.org_documentation_producerapi]

二，消息分发

1，kafka的分区副本概念

\* topic创建语句

sh ./bin/kafka-topics.sh --create --zookeeper 192.168.91.129:2181,192.168.91.129:2181,192.168.91.130:2181 --replication-factor 3 --partitions 3 --topic topicName

\* 分区概念解读

a，分区概况：每个 topic 可以划分多个分区（每个 Topic 至少有一个分 区），同一 topic 下的不同分区包含的消息是不同的。每个消息在被添加到分区时，都会被分配一个 offset（称之为偏 移量），它是消息在此分区中的唯一编号，kafka 通过 offset 保证消息在分区内的顺序，offset 的顺序不跨分区，即 kafka 只保证在同一个分区内的消息是有序的。该语句中被分为三个分区。

b，分区存储：topic分区是topic的实体概念，分区采用topic级别配置信息配置的轮循分区或者范围分区方式进行存储，遍历每一   个broker节点进行数据存储。一般情况下分区数等于broker个数，如果分区数和broker数都为3，则该topic在每一个broker上都会存在一个分区文件，命名分别为topicName-0\\1\\2，如果未指定分区个数，则会采用默认分区数

c，消息分发：消息是 kafka 中最基本的数据单元，在 kafka 中，一条消息 由 key、value 两部分构成，在发送一条消息时，我们可以 指定这个 key，那么 producer 会根据 key 和 partition 机 制来判断当前这条消息应该发送并存储到哪个 partition 中。 我们可以根据需要进行扩展 producer 的 partition 机制。key不存在时，会进行随机分发。

d，消息消费：消息分区用于对同一个groupId下的消费者进行绑定。在每一个groupId下，每一个消息分区会指定其中一个Consumer作为消费对象进行消息消费，若消息分区个数大于Consumer个数，则会存在部分Consumer消费多个消息分区；若消息分区个数消息Consumer个数，则多出部分的Consumer挂空。在不同的groupId下的Consumer，可以对同一条在该groupId下未被消息的消息进行二次消费。这样可以理解为，同一个groupId下，kafka队列类似于ActiveMQ的Queue队列，在不同groupId下，kafka队列类似于ActiveMQ的topic队列。

//消费指定分区的时候，不需要再订阅 
    //kafkaConsumer.subscribe(Collections.singletonList(topic)); 
    //消费指定的分区 
    TopicPartition topicPartition=new TopicPartition(topic,0); 
    kafkaConsumer.assign(Arrays.asList(topicPartition));

\* 副本概念解读

a，副本概况：我们已经知道Kafka的每个topic都可以分为多个Partition， 并且多个 partition 会均匀分布在集群的各个节点下。虽然 这种方式能够有效的对数据进行分片，但是对于每个 partition 来说，都是单点的，当其中一个 partition 不可用 的时候，那么这部分消息就没办法消费。所以 kafka 为了提高 partition 的可靠性而提供了副本的概念（Replica）,通 过副本机制来实现冗余备份。 每个分区可以有多个副本，并且在副本集合中会存在一个 leader 的副本，所有的读写请求都是由 leader 副本来进行 处理。剩余的其他副本都做为 follower 副本，follower 副 本 会 从 leader 副 本 同 步 消 息 日 志 。 这 个 有 点 类 似 zookeeper 中 leader 和 follower 的概念，但是具体的时间 方式还是有比较大的差异。所以我们可以认为，副本集会 存在一主多从的关系。 一般情况下，同一个分区的多个副本会被均匀分配到集群 中的不同 broker 上，当 leader 副本所在的 broker 出现故 障后，可以重新选举新的 leader 副本继续对外提供服务。 通过这样的副本机制来提高 kafka 集群的可用性。

b，副本存储：副本存储会根据副本算法存储在其他broker节点上，建议副本个数不多于broker个数。每一个消息分区会copy为指定副本个数的副本数量存储在其他broker节点上，如果副本个数和分区个数以及broker个数相等，则在每一个broker上都在存在全部分区的某一个副本，每一个分区的副本中会选择一个Leader副本存储上某一个broker上，用于客户端交互。其他副本作为follower副本用于数据备份。

c，副本同步：ISR&HW&LEO概念。

\* ISR 表示目前“可用且消息量与 leader 相差不多的副本集合， 这是整个副本集合的一个子集”。怎么去理解可用和相差不多 这两个词呢？具体来说，ISR 集合中的副本必须满足两个条件 1. 副本所在节点必须维持着与 zookeeper 的连接 2. 副本最后一条消息的 offset 与 leader 副本的最后一条消息 的 offset 之 间 的 差 值 不 能 超 过 指 定 的 阈 值 (replica.lag.time.max.ms) replica.lag.time.max.ms：如果该 follower 在此时间间隔 内一直没有追上过 leader 的所有消息，则该 follower 就 会被剔除 isr 列表

\* HW(HighWatermark)和 LEO(Log End Offset) ：这两个参 做技术人的指路明灯，做职场生涯的精神导师 数跟 ISR 集合紧密关联。HW 标记了一个特殊的 offset，当 消费者处理消息的时候，只能拉去到 HW 之前的消息，HW 之后的消息对消费者来说是不可见的。也就是说，取 partition 对应 ISR 中最小的 LEO 作为 HW，consumer 最 多只能消费到 HW 所在的位置。每个 replica 都有 HW， leader 和 follower 各自维护更新自己的 HW 的状态。一条 消息只有被 ISR 里的所有 Follower 都从 Leader 复制过去 才会被认为已提交。这样就避免了部分数据被写进了 Leader，还没来得及被任何 Follower 复制就宕机了，而造 成数据丢失（Consumer 无法消费这些数据）。而对于 Producer 而言，它可以选择是否等待消息 commit，这可 以通过 acks 来设置。这种机制确保了只要 ISR 有一个或以 上的 Follower，一条被 commit 的消息就不会丢失。

三，消息消费

1，消费者的分区分配机制

a，如前所述，同一个groupId下，多个消息者会分别去消费多个分区。多消费者过多，则多余消费者挂起；若分区过多，则部分消费者消费多个分区。

b，不同griupId下，分区内信息会被所分配到的consumer进行多次消费。

2，消费者消费位置标识 -- offset

a，offset概述：每个 topic可以划分多个分区（每个 Topic 至少有一个分区），同一 topic 下的不同分区包含的消息是不同的。每个消息在被添 加到分区时，都会被分配一个 offset（称之为偏移量），它 是消息在此分区中的唯一编号，kafka 通过 offset 保证消息 在分区内的顺序，offset 的顺序不跨分区，即 kafka 只保证 在同一个分区内的消息是有序的； 对于应用层的消费来说， 每次消费一个消息并且提交以后，会保存当前消费到最 近的一个 offset。

b，offset存储：每一个用户的offset偏移量保存在\_\_consumer\_offsets-\*中, \*根据groupId的hashCode%50求出。

c，查看偏移量命令，从截图可以看出，当前消息发送到不同的broker上，当前消息者消费全部的broker并移动偏移量

sh kafka-simple-consumer-shell.sh --topic __consumer_offsets --partition 3 --broker-list 192.168.91.128:9092,192.168.91.129:9092,192.168.91.130:9092 --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter"

![70][]

四，消息存储

1，消息文件存储机制

a，日志存储在log.dirs配置路径下，具体分区消息日志存储在topicName-partitionId文件下

![70 1][]

![70 2][]

![70 3][]

b，日志文件命名方式：上一个文件最后一个消息的offset+1

c，索引文件与日志文件映射：offset表示索引迁移量，position表示物理偏移量

d，时间索引文件与日志文件映射：根据时间戳映射索引便宜量

e，日志文件查看，索引文件和时间戳文件查看方式相同

sh kafka-run-class.sh kafka.tools.DumpLogSegments --files /tmp/kafka-logs/topicName-2/00000000000000000000.log --print-data-log

\* 日志查看

![70 4][]

\* 索引查看

![70 5][]

\* 时间戳查看

![70 6][]

2，日志清除及压缩策略

a，日志清除：通过 log.retention.bytes 和 log.retention.hours 这两个参 数来设置，当其中任意一个达到要求，都会执行删除。

b，日志压缩：按照系统运行逻辑，消费者只关系当前key对应的最新value值，而历史value值不在消费者的关注范围内。因 此，我们可以开启 kafka 的日志压缩功能，服务端会在后 台启动启动Cleaner 线程池，定期将相同的 key 进行合并， 只保留最新的 value 值。

[http_kafka.apache.org_documentation_producerapi]: http://kafka.apache.org/documentation/#producerapi
[70]: /images/20220509/338e704132d44a4b99f6d9b7605bb4c1.png
[70 1]: /images/20220509/a3932e82cbe0487d905f3d0e6c0979ee.png
[70 2]: /images/20220509/6cc4b95db6814a49bd11bd0be748145b.png
[70 3]: /images/20220509/c06f25ebb5204aafaea49af79c6c0a34.png
[70 4]: /images/20220509/a7c71d623e3a4776afe1a888f01d0bcd.png
[70 5]: /images/20220509/1263e272d36c42d298093f3836362d00.png
[70 6]: /images/20220509/cc3cbed6c4094e62904bd3f56e88db90.png