浅谈MQ-kafka基本操作

我会带着你远行 2023-01-21 10:26 25阅读 0赞

# kafka的基本操作 #

这里以docker部署的kafka为例

# 如果是使用wurstmeister/kafka镜像
    docker exec -it <containerID> /bin/sh
    cd /opt/kafka  # kafka 安装目录
    cd /kafka # topic消息记录文件目录

## 创建topic ##

sh kafka-topics.sh --create --zookeeper 192.168.1.40:2181 --replication-factor 3 --partitions 3 --topic test
    
    # Replication-factor 表示该topic需要在不同的broker中保存几份，这里设置成1，表示在两个broker中保存两份
    # Partitions 分区数

## 查看topic ##

# sh kafka-topics.sh --list --zookeeper 192.168.1.40:2181
    
    LISTEN_IM_STATUS
    __consumer_offsets
    group_live
    group_live_message
    test

## 查看topic属性 ##

# sh kafka-topics.sh --describe --zookeeper 192.168.1.40:2181 --topic group_live_message
    
    Topic: group_live_message       PartitionCount: 3(分区数)       ReplicationFactor: 1(复制因子)    Configs: 
            Topic: group_live_message       Partition: 0    Leader: 1       Replicas: 1     Isr: 1
            Topic: group_live_message       Partition: 1    Leader: 2       Replicas: 2     Isr: 2
            Topic: group_live_message       Partition: 2    Leader: 0       Replicas: 0     Isr: 0
                                            (分区)          (1个主节点)     (1个副本)

## 修改分区数 ##

# sh kafka-topics.sh --alter --zookeeper 192.168.1.40:2181 --topic group_live_message --partitions 3
    
    WARNING: If partitions are increased for a topic that has a key, the partition logic or ordering of the messages will be affected
    Adding partitions succeeded!

## 修改副本数（复制因子） ##

# 1.不能通过以下方法实现更改副本数
    # sh kafka-topics.sh --alter --zookeeper 192.168.1.40:2181 --topic group_live_message --replication-factor 3
    
    Option "[replication-factor]" can't be used with option "[alter]"
    Option                                   Description                            
    ------                                   -----------                            
    --alter                                  Alter the number of partitions,
    ...
    
    # 2.正确方式：https://kafka.apache.org/documentation/#basic_ops_increase_replication_factor
    在/opt/kafka/bin 创建一个文件increase-replication-factor.json
    
    文间内容：
    {"version":1,
    "partitions":[
    {"topic":"group_live_message","partition":0,"replicas":[0,1,2]},
    {"topic":"group_live_message","partition":1,"replicas":[0,1,2]},
    {"topic":"group_live_message","partition":2,"replicas":[0,1,2]}
    ]}
    
    # 3.执行脚本
    # sh kafka-reassign-partitions.sh -zookeeper 192.168.1.40:2181 --reassignment-json-file increase-replication-factor.json --execute
    
    Current partition replica assignment
    
    {"version":1,"partitions":[{"topic":"group_live_message","partition":2,"replicas":[0],"log_dirs":["any"]},{"topic":"group_live_message","partition":1,"replicas":[2],"log_dirs":["any"]},{"topic":"group_live_message","partition":0,"replicas":[1],"log_dirs":["any"]}]}
    
    Save this to use as the --reassignment-json-file option during rollback
    Successfully started reassignment of partitions.
    
    # 4.查看更新成功
    # sh kafka-topics.sh --describe --zookeeper 192.168.1.40:2181 --topic group_live_message
    
    Topic: group_live_message       PartitionCount: 3       ReplicationFactor: 3    Configs: 
            Topic: group_live_message       Partition: 0    Leader: 1       Replicas: 0,1,2 Isr: 1,0,2
            Topic: group_live_message       Partition: 1    Leader: 2       Replicas: 0,1,2 Isr: 2,1,0
            Topic: group_live_message       Partition: 2    Leader: 0       Replicas: 0,1,2 Isr: 0,2,1

## 订阅topic（消费信息） ##

# sh kafka-console-consumer.sh --bootstrap-server 192.168.1.40:9092 --topic group-live-message --from-beginning

## 生产者（发送消息） ##

# sh kafka-console-producer.sh --broker-list 192.168.1.40:9092 --topic group-live-message

## 删除topic ##

### step1： ###

如果需要被删除topic 此时正在被程序 produce和consume，则这些生产和消费程序需要停止。因为如果有程序正在生产或者消费该topic，则该topic的offset信息一致会在broker更新。调用kafka delete命令则无法删除该topic。同时，需要设置 `auto.create.topics.enable = false`，默认设置为true。如果设置为true，则produce或者fetch 不存在的topic也会自动创建这个topic。这样会给删除topic带来很多意向不到的问题。所以，这一步很重要，必须设置`auto.create.topics.enable = false`，并认真把生产和消费程序彻底全部停止。

### step2： ###

server.properties设置`delete.topic.enable=true`如果没有设置`delete.topic.enable=true`，则调用kafka 的delete命令无法真正将topic删除，而是显示（marked for deletion）

### step3： ###

调用命令删除topic：

sh kafka-topics.sh --delete --zookeeper 192.168.1.40:9092 --topic group_live_message

### step4： ###

删除kafka存储目录（server.properties文件`log.dirs`配置，默认为"/data/kafka-logs"）相关topic的数据目录。

注意：如果kafka 有多个 broker，且每个broker 配置了多个数据盘（比如`/data/kafka-logs,/data1/kafka-logs...`），且topic也有多个分区和replica，则需要对所有broker的所有数据盘进行扫描，删除该topic的所有分区数据。一般而言，经过上面4步就可以正常删除掉topic和topic的数据。但是，如果经过上面四步，还是无法正常删除topic，则需要对kafka在zookeeer的存储信息进行删除。具体操作如下：  
（注意：以下步骤里面，kafka在zk里面的节点信息是采用默认值，如果你的系统修改过kafka在zk里面的节点信息，则需要根据系统的实际情况找到准确位置进行操作）

### step5： ###

找一台部署了zk的服务器，使用命令：

bin/zkCli.sh -server 192.168.1.40:9092

登录到zk shell，然后找到topic所在的目录：`ls /brokers/topics`，找到要删除的topic，然后执行命令：

deleteall /brokers/topics/[topic name]

即可，此时topic被彻底删除。如果topic 是被标记为 marked for deletion，则通过命令 ls /admin/delete\_topics，找到要删除的topic，然后执行命令：

deleteall /admin/delete_topics/[topic name]

备注：

网络上很多其它文章还说明，需要删除topic在zk上面的消费节点记录、配置节点记录，比如：

deleteall /consumers/[consumer-group]
    deleteall /config/topics/[topic name]

其实正常情况是不需要进行这两个操作的，如果需要，那都是由于操作不当导致的。比如step1停止生产和消费程序没有做，step2没有正确配置。也就是说，正常情况下严格按照step1 – step5 的步骤，是一定能够正常删除topic的。

### step6： ###

完成之后，调用命令：

./bin/kafka-topics.sh --list --zookeeper [zookeeper server:port]

查看现在kafka的topic信息。正常情况下删除的topic就不会再显示。但是，如果还能够查询到删除的topic，则重启zk和kafka即可。

## kafka消息保留配置 ##

配置在`/bin/server.properties`文件中

*  `log.retention.ms` 消息时间
    
    Kafka通常根据时间决定数据可以保留多久。默认使用log.retention.hours参数配置时间，默认值是168小时，也就是一周。除此之外，还有其他两个参数，`log.retention.minutes`和`log.retention.ms`，这三个参数作用是一样的，都是决定消息多久以会被删除，不过还是推荐使用`log.retention.ms`，如果指定了不止一个参数，Kafka会优先使用最小值的那个参数。
 *  `log.retention.bytes` 消息大小
    
    通过保留的消息字节数来判断小是否过期，它的值通过参数`log.retention.bytes`来指定，作用在每一个分区上，也就是说如果一个包含8个分区的主题，并且`log.retention.bytes`被设置为1GB，那么这个主题最多可以保留8GB的数据，所以，当主题的分区个数增加时，整个主题可以保留的数据也随之增加。

> 注意：如果同时指定了两个参数没只要任意一个参数得到满足，消息就会被删除。例如，假设log.retention.ms为86400000（也就是一天），log.retention.bytes的值设置为1GB，如果消息字节总数在不到一天的时间就超过了1GB，那么堆出来的部分就会被删除，相反，如果消息字节总数小与1GB，那么一天之后这些消息也会被删除，尽管分区的数据总量小于1GB

*  `log.segment.bytes` 日志片段大小
    
    当消息来到broker时，它们就会被追加到分区的当前日志片段上，当日志片段大小到达`log.segment.bytes`指定的上限（默认是1GB）时，当前日志片段就会被关闭，一个新的日志片段就会被打开。如果一个日志之片段被关闭，就开始等待过期时间。这个参数的值越小们就会越频繁的关闭和分配新文件，从而降低了磁盘写入的整体效率。
 *  `log.segment.ms` 日志片段时间
    
    指定了多长时间之后日志片段会被关闭，就像`log.retention.bytes`和`log.retention.ms`这两个参数一样。`log.segment.bytes`和`log.segment.ms`这两个参数之间也不存在互斥问题。日志片段会在大小或时间达到上限时被关闭，就看哪个条件晓得到满足。默认情况下`log.segment.ms`没有设定值，所以只根据大小来关闭日志片段
 *  `message.max.bytes` 单条消息大小
    
    broker通过设置`message.max.bytes`参数来限制单个消息的大小，默认值时1000000，也就是1MB。如果生产者尝试发送的消息超过1MB，不仅消息不会被接受，还会受到broker返回的错误信息。跟其他与字节相关的配置参数一样，该参数指的是压缩后的消息大小，也就是说，只要压缩后的消息小于`message.max.bytes`指定的值，消息的实际大小可以远大于这个值。

## kafka集群状态(访问zookeeper查看信息) ##

![image][]

# window连接zookeeper
    ./zkCli.cmd -server ip:port 
    
    # 查看broker存活情况
    ls /brokers/ids

## Q&A ##

当kafka集群宕机，或者集群中其中一台机器系统重新安装后部署kafka时，我们就要注意启动kafka时的日志情况。`docker logs kafka`

1.  Error while executing topic command : replication factor: 3 larger than available brokers: 0

这个问题是kafka有部分broker实例没有加入到kafka集群中，需要通过连接zookeeper，执行 `ls /brokers/ids` 查看有几台机器没有被注册上来。

1.  The Cluster ID xxx doesn’t match stored clusterId Some(xxx) in meta.properties. The broker is trying

意思是集群id跟元数据meta.properties中存储的不一致，导致启动失败。因此去查看meta.properties文件中的元数据信息。这个文件的存储路径是通过/config/server.properties配置文件中的log.dirs属性配置的。所以通过配置文件找到meta.properties，修改里面的cluster.id或者删除meta.properties 即可。如修改cluster.id重新部署也不行，则把meta.properties文件删除。

# Kafka使用方法 #

## Java中使用kafka进行通信 ##

### 依赖 ###

<dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>2.0.0</version>
    </dependency>

### 发送端代码 ###

import com.sendbp.eduz.kafka.Kafka4JTest;
    import org.apache.kafka.clients.producer.*;
    import org.apache.kafka.common.serialization.IntegerSerializer;
    import org.apache.kafka.common.serialization.StringSerializer;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import java.util.Optional;
    import java.util.Properties;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.TimeUnit;
    
    /** * 生产者 * * @author JC.Lin * @date 2020-06-24 16:22 */
    public class Producer extends Thread { 
        private static final Logger logger = LoggerFactory.getLogger(Kafka4JTest.class);
    
        private final KafkaProducer<Integer, String> producer;
        private final String topic;
    
        public Producer(String topic) { 
            Properties properties = new Properties();
            properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "<外网IP>:9092,<外网IP>:9095,<外网IP>:9096");
            properties.put(ProducerConfig.CLIENT_ID_CONFIG, "practice-producer");
            properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class.getName());
            properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    
            // batch.size: 生产者发送多个消息到broker上的同一个分区时，为了减少网络请求带来的性能开销，通过批量的方式
            // 来提交消息，可以通过这个参数来控制批量提交的字节数大小，默认大小是16384byte,也就是16kb，
            // 意味着当一批消息大小达到指定的batch.size的时候会统一发送
            properties.put(ProducerConfig.BATCH_SIZE_CONFIG, "16kb");
            // linger.ms: Producer默认会把两次发送时间间隔内收集到的所有Requests进行一次聚合然后再发送，以此提高吞
            // 吐量，而linger.ms就是为每次发送到broker的请求增加一些delay，以此来聚合更多的Message请求。
            // 这个有点想TCP里面的Nagle算法，在TCP协议的传输中，为了减少大量小数据包的发送，采用了Nagle
            // 算法，也就是基于小包的等-停协议
            properties.put(ProducerConfig.LINGER_MS_CONFIG, "500");
    
            producer = new KafkaProducer<>(properties);
            this.topic = topic;
        }
    
        @Override
        public void run() { 
            int num = 0;
            while (num < 50) { 
                String msg = "pratice test message:" + num;
                try { 
                    // 同步方式
    // producer.send(new ProducerRecord<>(topic, msg)).get();
    
                    // 异步方式
                    producer.send(new ProducerRecord<>(topic, msg), (recordMetadata, e) -> { 
                        System.out.println("callback: " + recordMetadata.offset() + "->" + recordMetadata.partition());
                    });
    
                    TimeUnit.SECONDS.sleep(2);
                    num++;
                } catch (Exception e) { 
                    logger.error("", e);
                }
            }
    
        }
    
        public static void main(String[] args) { 
            new Producer("test").start();
        }
    
    }

### 接收端代码 ###

import org.apache.kafka.clients.consumer.ConsumerConfig;
    import org.apache.kafka.clients.consumer.ConsumerRecords;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
    import org.apache.kafka.common.serialization.IntegerDeserializer;
    import org.apache.kafka.common.serialization.StringDeserializer;
    
    import java.time.Duration;
    import java.util.Collections;
    import java.util.Properties;
    
    /** * 消费者 * @author JC.Lin * @date 2020-06-24 16:23 */
    public class Consumer extends Thread { 
    
        private final KafkaConsumer<Integer, String> consumer;
        private final String topic;
    
        public Consumer(String topic) { 
            Properties properties = new Properties();
            properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "<外网IP>:9092,<外网IP>:9095,<外网IP>:9096");
            properties.put(ConsumerConfig.GROUP_ID_CONFIG, "practice-consumer");
            properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");//设置offset自动提交
            properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");//自动提交间隔时间
            properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
            properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class.getName());
            properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
            properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");//对于当前groupid来说，消息的offset从最早的消息开始消费
            consumer = new KafkaConsumer<>(properties);
            this.topic = topic;
        }
    
        @Override
        public void run() { 
            while (true) { 
                consumer.subscribe(Collections.singleton(this.topic));
                ConsumerRecords<Integer, String> records = consumer.poll(Duration.ofSeconds(1));
                records.forEach(record -> { 
                    System.out.println(record.key() + " " + record.value() + " -> offset:" + record.offset());
                });
            }
        }
    
        public static void main(String[] args) { 
            new Consumer("test").start();
        }
    
    }

### batch.size ###

生产者发送多个消息到broker上的同一个分区时，为了减少网络请求带来的性能开销，通过批量的方式来提交消息，可以通过这个参数来控制批量提交的字节数大小，默认大小是16384byte,也就是16kb，意味着当一批消息大小达到指定的batch.size的时候会统一发送

### linger.ms ###

Producer默认会把两次发送时间间隔内收集到的所有Requests进行一次聚合然后再发送，以此提高吞吐量，而linger.ms就是为每次发送到broker的请求增加一些delay，以此来聚合更多的Message请求。这个有点想TCP里面的Nagle算法，在TCP协议的传输中，为了减少大量小数据包的发送，采用了Nagle算法，也就是基于小包的等-停协议。

> batch.size和linger.ms这两个参数是kafka性能优化的关键参数，会发现batch.size和linger.ms这两者的作用是一样的，如果两个都配置了，那么怎么工作的呢？实际上，当二者都配置的时候，只要满足其中一个要求，就会发送请求到broker上

### group.id ###

`consumer group`是kafka提供的可扩展且具有容错性的消费者机制。既然是一个组，那么组内必然可以有多个消费者或消费者实例(consumer instance)，它们共享一个公共的ID，即group ID。组内的所有消费者协调在一起来消费订阅主题(subscribed topics)的所有分区(partition)。当然，**每个分区只能由同一个消费组内的一个consumer来消费**.

如下图所示，分别有三个消费者，属于两个不同的group，那么对于firstTopic这个topic来说，这两个组的消费者都能同时消费这个topic中的消息，对于此事的架构来说，这个firstTopic就类似于ActiveMQ中的topic概念。

![image][image 1]

不同的组的消费者可以在同一时间接收到相同的信息

如下图所示，如果3个消费者都属于同一个group，那么此事firstTopic就是一个Queue的概念

![image][image 2]

同一个组内的消费者，同一时间下一个消息只能由一个消费者消费。

### enable.auto.commit ###

消费者消费消息以后自动提交，只有当消息提交以后，该消息才不会被再次接收到，还可以配合  
`auto.commit.interval.ms`控制自动提交的频率。

当然，我们也可以通过`consumer.commitSync()`的方式实现手动提交。

### auto.offset.reset ###

这个参数是针对新的groupid中的消费者而言的，当有新groupid的消费者来消费指定的topic时，对于该参数的配置，会有不同的语义

*  auto.offset.reset=latest情况下，新的消费者将会从其他消费者最后消费的offset处开始消费Topic下的  
    消息
 *  auto.offset.reset=earliest情况下，新的消费者会从该topic最早的消息开始消费
 *  auto.offset.reset=none情况下，新的消费者加入以后，由于之前不存在offset，则会直接抛出异常。

### max.poll.records ###

此设置限制每次调用poll返回的消息数，这样可以更容易的预测每次poll间隔要处理的最大值。通过调整此值，可以减少poll间隔

## Springboot+kafka ##

springboot的版本和kafka的版本，有一个对照表格，如果没有按照正确的版本来引入，那么会存在版本问题导致ClassNotFound的问题，具体请参考

https://spring.io/projects/spring-kafka

![image][image 3]

### 依赖 ###

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

### 配置 ###

springboot-kafka自动装配类：org.springframework.boot.autoconfigure.kafka.KafkaProperties

spring: 
      kafka:
        bootstrapservers: <公网IP>:9092,<公网IP>:9095,<公网IP>:9096
    
        producer:
          keyserializer: org.apache.kafka.common.serialization.StringSerializer
          valueserializer: org.apache.kafka.common.serialization.StringSerializer
    
        consumer:
          group-id: test-consumer-group
          auto-offset-reset: earliest
          enable-auto-commit: true
          keydeserializer: org.apache.kafka.common.serialization.StringDeserializer
          valuedeserializer: org.apache.kafka.common.serialization.StringDeserializer

### KafkaProducer ###

@Component
    public class KafkaProducer { 
        @Autowired
        private KafkaTemplate<String,String> kafkaTemplate;
        
        public void send(){ 
            kafkaTemplate.send("test","msgKey","msgData");
        }
    }

### KafkaConsumer ###

@Component
    public class KafkaConsumer { 
        @KafkaListener(topics = { "test"})
        public void listener(ConsumerRecord record){ 
            Optional<?> msg=Optional.ofNullable(record.value());
            if(msg.isPresent()){ 
                System.out.println(msg.get());
            }
        }
    }

### Test ###

import org.junit.AfterClass;
    import org.junit.runner.RunWith;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.test.context.ActiveProfiles;
    import org.springframework.test.context.junit4.SpringRunner;
    
    import java.io.IOException;
    import java.io.OutputStream;
    import java.io.PrintStream;
    
    @RunWith(SpringRunner.class)
    @SpringBootTest(classes = KafkaApplication.class)
    @ActiveProfiles("dev")
    public abstract class TestBase { 
    
        /** * 在所有测试运行完毕后关闭控制台输出，因为Spring Netflix内部Bug会在测试运行完毕后结束进程时产生与测试无关的异常 */
        @AfterClass
        public static void ignoreConsolePrint(){ 
            OutputStream ignoreOutputStream = new OutputStream() { 
                @Override
                public void write(int b) throws IOException { 
                    //ignore
                }
            };
            System.setOut(new PrintStream(ignoreOutputStream));
            System.setErr(new PrintStream(ignoreOutputStream));
        }
    
    }

import com.sendbp.eduz.TestBase;
    import org.junit.Test;
    
    import javax.annotation.Resource;
    
    /** * @author JC.Lin * @date 2020-06-26 18:14 */
    public class SpringBootKafkaTest extends TestBase { 
    
        @Resource
        private KafkaProducer kafkaProducer;
    
        @Test
        public void Test1() { 
            for (int i = 0; i < 3; i++) { 
                kafkaProducer.send();
                try { 
                    Thread.sleep(3000);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
        }
    
    }

[image]: /images/20221020/bebc3f65c9494654af3c53c5e6169941.png
[image 1]: /images/20221020/2ef33e96ae88474cb4845a658f99b4dc.png
[image 2]: /images/20221020/d3878740b91e4bda83b55a49c2533fcd.png
[image 3]: /images/20221020/6edff753e3864326b53dd1664e1e2de8.png