消息队列MQ

kafka消费者接收分区测试

本文主要是介绍kafka消费者接收分区测试,对大家解决编程问题具有一定的参考价值,需要的程序猿们随着小编来一起学习吧!

【README】

  • 本文演示了当有新消费者加入组后,其他消费者接收分区情况;
  • 本文还模拟了 broker 宕机的情况;
  • 本文使用的是最新的 kafka3.0.0 ;
  • 本文测试案例,来源于 消费者接收分区的5种模型,建议先看模型,refer2   https://blog.csdn.net/PacosonSWJTU/article/details/121853461icon-default.png?t=LA92https://blog.csdn.net/PacosonSWJTU/article/details/121853461

【1】kafka测试环境准备

1)kafka集群 

  • 3个broker,分别为 centos201, centos202, centos203 ,id分别为 1,2,3 ;
  • topic, 3个分区,2个副本;

 2)生产者代码;

public class MyProducer {
    public static void main(String[] args) {
        /* 1.创建kafka生产者的配置信息 */
        Properties props = new Properties();
        /*2.指定连接的kafka集群, broker-list */
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.163.201:9092,192.168.163.202:9092,192.168.163.203:9092");
        /*3.ack应答级别*/
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        /*4.重试次数*/
        props.put(ProducerConfig.RETRIES_CONFIG, 0);
        /*5.批次大小,一次发送多少数据,当数据大于16k,生产者会发送数据到 kafka集群 */
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16 * KfkNumConst._1K);
        /*6.等待时间, 等待时间超过1毫秒,即便数据没有大于16k, 也会写数据到kafka集群 */
        props.put(ProducerConfig.LINGER_MS_CONFIG, 1);

        // 超时时间
        props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 3000);
        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 3000);

        /*7. RecordAccumulator 缓冲区大小*/
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 32 * KfkNumConst._1M);
        /*8. key, value 的序列化类 */
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        /** 设置压缩算法 */
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");
        /** 设置拦截器 */
//        props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, Arrays.asList(TimeInterceptor.class.getName()));
        /** 设置阻塞超时时间 */
        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 3600 * 1000);

        /* 9.创建生产者对象 */
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        /* 10.发送数据 */
        int order = 1;
        for (int i = 0; i < 100000; i++) {
            for (int j = 0; j < 3; j++) {
                Future<RecordMetadata> future = producer.send(new ProducerRecord<String, String>("hello10",j, "", String.format("[%s] ", order++) + " > " + DataFactory.INSTANCE.genChar(5)));
                try {
                    System.out.println("[生产者] 分区【" + future.get().partition() + "】-offset【" + future.get().offset() + "】");
                } catch (Exception e) {
                }
            }
            try {
                TimeUnit.MILLISECONDS.sleep(500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        /* 11.关闭资源 */
        producer.close();
        System.out.println("kafka生产者写入数据完成");
    }
}

生产者,会向每个分区发送1条消息,发送完成后,睡眠500ms; 共计循环 10w次; 共计5w秒;计划耗时 10+小时;(这里其他同学可以自行设置为其他值)

3)4个消费者;编号为1,2,3,4

public class MyConsumer1 {
	public static void main(String[] args) {
		/* 1.创建消费者配置信息 */
		Properties props = new Properties();
		/*2.给配置信息赋值*/
		/*2.1连接的集群*/
		props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "centos201:9092,centos202:9092,centos203:9092");
		/*2.2开启自动提交 */
		props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
		/*2.3 自动提交的间隔时间*/
		props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
		/*2.4 key value的反序列化 */
		props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
		props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
		/*2.5 消费者组 */
		props.put(ConsumerConfig.GROUP_ID_CONFIG, "hello10G1"); // group.id
		/*2.6 重置消费者的offset */ 
		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest"); // 默认值是 lastest
		/*2.7 关闭自动提交 */
		props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

		/* 创建消费者 */
		KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props); 
		/* 订阅主题 */
		consumer.subscribe(Arrays.asList("hello10"));

		/* 指定消费者的每个分区从偏移量1开始读取,下面的poll方法就会从位置1开始消费消息  */
//		for (TopicPartition partition : consumer.assignment()) {
//			consumer.seek(partition, 1);
//		}

		// 消费消息
		try {
			// 死循环
			while(!Thread.interrupted()) {
				try {
					System.out.println(DateUtils.getNowTimestamp() + " 消费者1-等待消费消息");
					TimeUnit.MILLISECONDS.sleep(100);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				// 消费消息-获取数据
				ConsumerRecords<String, String> consumerRds  = consumer.poll(100);
				// 遍历 ConsumerRecords
				for(ConsumerRecord<String, String> rd : consumerRds) {
					System.out.println("消费者1-分区【" + rd.partition() + "】offset【" + rd.offset() + "】 -> " + DateUtils.getNowTimestamp() + rd.key() + "--" + rd.value());
				}
				consumer.commitSync(); // 同步提交
			}
		} finally {
			// 记得关闭消费者
			consumer.close();
		}
	}
}

这样的消费者有4个,分别编号为 消费者 1,2,3,4 ;我的意思是4个不同的消费者类,以便打印日志标识;

我的消费者消费的是 lastest 最新产生的消费,这里可以自行设置为其他值,如 earlies;

4)添加日志配置,不打印 debug日志(因为kafka消费者debug日志很多)

新建 logback.xml ,设置仅打印info以上级别日志;

<?xml version="1.0" encoding="UTF-8"?>
<configuration>
    <logger name="org.apache.kafka.clients" level="info" />
</configuration>

5)为了直观展示消费详情,我会用命令行启动4个不同消费者,而用idea启动生产者;但编译都通过maven;


【2】kafka测试

【2.1】测试1:当有新消费者加入后,整个消费者组成员接收分区情况; 

写在前面: 文末会po出命令行启动消费者的命令及参数;

消费者接收分区消息模型,参见

step0)启动 生产者,发送消息到kafka;

step2)命令行启动消费者1,消息消费日志如下:

消费者1接收了3个分区消息; 

 step2)命令行启动消费者2,群组消费日志如下:

消费者1接收了个分区2消息;

消费者2接收了分区0和分区2的消息;

  step3)命令行继续启动消费者3,群组消费日志如下:

消费者1接收了个分区2消息;
消费者2接收了分区0的消息;
消费者3接收了分区1的消息;

 step4)命令行继续启动消费者4, 日志如下:

消费者1接收了个分区2消息;
消费者2接收了分区0的消息;
消费者3接收了分区1的消息;

消费者4空闲;

 


【2】 模拟kafka broker 宕机

写在前面,模拟宕机前先查看 topic 详情

(图1)

step1) 停止掉 201 broker的服务

情况1:topic的分区没有受影响,但leader 副本选举为3,比较本图和图1,看差别; 

 情况2:所有消费者全部阻塞,直到超时全部抛出异常;

等待 kafka集群的控制器,首领副本选择完成后,又可以接收消费者请求; 

  • 补充1: 这里有一小段时间延时,即当有broker宕机后,需要重新选举控制器,首领副本等;而且会发生分区再均衡;

 step2)重启 201;消费日志:如下:

消费者1接收了个分区1消息;
消费者2接收了分区2的消息;
消费者3空闲;
消费者4接收了分区0的消息;

 之所以 消费者3空闲,消费者4忙碌,是因为 broker 动态上下线,导致了分区再均衡使得分区所有权从消费者A转到消费者B(201宕机前,是消费者3忙碌,消费者4空闲);

【小结】

1,要保证kafka消息可靠性,需要 生产者,broker,消费者3方的全力配合;

2,本文这里仅记录了一部分 kafka集群异常的情况;


【附录】

命令行启动消费者命令及参数;仅供参考;因为路径肯定不一样;

其实,这是拷贝idea的执行日志里的命令,如下:

 

java -classpath D:\Java\jdk1.8.0_172\jre\lib\charsets.jar;D:\Java\jdk1.8.0_172\jre\lib\deploy.jar;D:\Java\jdk1.8.0_172\jre\lib\ext\access-bridge-64.jar;D:\Java\jdk1.8.0_172\jre\lib\ext\cldrdata.jar;D:\Java\jdk1.8.0_172\jre\lib\ext\dnsns.jar;D:\Java\jdk1.8.0_172\jre\lib\ext\jaccess.jar;D:\Java\jdk1.8.0_172\jre\lib\ext\jfxrt.jar;D:\Java\jdk1.8.0_172\jre\lib\ext\localedata.jar;D:\Java\jdk1.8.0_172\jre\lib\ext\nashorn.jar;D:\Java\jdk1.8.0_172\jre\lib\ext\sunec.jar;D:\Java\jdk1.8.0_172\jre\lib\ext\sunjce_provider.jar;D:\Java\jdk1.8.0_172\jre\lib\ext\sunmscapi.jar;D:\Java\jdk1.8.0_172\jre\lib\ext\sunpkcs11.jar;D:\Java\jdk1.8.0_172\jre\lib\ext\zipfs.jar;D:\Java\jdk1.8.0_172\jre\lib\javaws.jar;D:\Java\jdk1.8.0_172\jre\lib\jce.jar;D:\Java\jdk1.8.0_172\jre\lib\jfr.jar;D:\Java\jdk1.8.0_172\jre\lib\jfxswt.jar;D:\Java\jdk1.8.0_172\jre\lib\jsse.jar;D:\Java\jdk1.8.0_172\jre\lib\management-agent.jar;D:\Java\jdk1.8.0_172\jre\lib\plugin.jar;D:\Java\jdk1.8.0_172\jre\lib\resources.jar;D:\Java\jdk1.8.0_172\jre\lib\rt.jar;D:\workbench_idea\study4vw\vwstudy22\target\classes;D:\software_cluster\mvn_repo\.m2\repository\org\springframework\boot\spring-boot-starter-web\2.5.4\spring-boot-starter-web-2.5.4.jar;D:\software_cluster\mvn_repo\.m2\repository\org\springframework\boot\spring-boot-starter\2.5.4\spring-boot-starter-2.5.4.jar;D:\software_cluster\mvn_repo\.m2\repository\org\springframework\boot\spring-boot\2.5.4\spring-boot-2.5.4.jar;D:\software_cluster\mvn_repo\.m2\repository\org\springframework\boot\spring-boot-autoconfigure\2.5.4\spring-boot-autoconfigure-2.5.4.jar;D:\software_cluster\mvn_repo\.m2\repository\org\springframework\boot\spring-boot-starter-logging\2.5.4\spring-boot-starter-logging-2.5.4.jar;D:\software_cluster\mvn_repo\.m2\repository\ch\qos\logback\logback-classic\1.2.5\logback-classic-1.2.5.jar;D:\software_cluster\mvn_repo\.m2\repository\ch\qos\logback\logback-core\1.2.5\logback-core-1.2.5.jar;D:\software_cluster\mvn_repo\.m2\repository\org\apache\logging\log4j\log4j-to-slf4j\2.14.1\log4j-to-slf4j-2.14.1.jar;D:\software_cluster\mvn_repo\.m2\repository\org\apache\logging\log4j\log4j-api\2.14.1\log4j-api-2.14.1.jar;D:\software_cluster\mvn_repo\.m2\repository\org\slf4j\jul-to-slf4j\1.7.32\jul-to-slf4j-1.7.32.jar;D:\software_cluster\mvn_repo\.m2\repository\jakarta\annotation\jakarta.annotation-api\1.3.5\jakarta.annotation-api-1.3.5.jar;D:\software_cluster\mvn_repo\.m2\repository\org\springframework\spring-core\5.3.9\spring-core-5.3.9.jar;D:\software_cluster\mvn_repo\.m2\repository\org\springframework\spring-jcl\5.3.9\spring-jcl-5.3.9.jar;D:\software_cluster\mvn_repo\.m2\repository\org\yaml\snakeyaml\1.28\snakeyaml-1.28.jar;D:\software_cluster\mvn_repo\.m2\repository\org\springframework\boot\spring-boot-starter-json\2.5.4\spring-boot-starter-json-2.5.4.jar;D:\software_cluster\mvn_repo\.m2\repository\com\fasterxml\jackson\core\jackson-databind\2.12.4\jackson-databind-2.12.4.jar;D:\software_cluster\mvn_repo\.m2\repository\com\fasterxml\jackson\core\jackson-annotations\2.12.4\jackson-annotations-2.12.4.jar;D:\software_cluster\mvn_repo\.m2\repository\com\fasterxml\jackson\core\jackson-core\2.12.4\jackson-core-2.12.4.jar;D:\software_cluster\mvn_repo\.m2\repository\com\fasterxml\jackson\datatype\jackson-datatype-jdk8\2.12.4\jackson-datatype-jdk8-2.12.4.jar;D:\software_cluster\mvn_repo\.m2\repository\com\fasterxml\jackson\datatype\jackson-datatype-jsr310\2.12.4\jackson-datatype-jsr310-2.12.4.jar;D:\software_cluster\mvn_repo\.m2\repository\com\fasterxml\jackson\module\jackson-module-parameter-names\2.12.4\jackson-module-parameter-names-2.12.4.jar;D:\software_cluster\mvn_repo\.m2\repository\org\springframework\boot\spring-boot-starter-tomcat\2.5.4\spring-boot-starter-tomcat-2.5.4.jar;D:\software_cluster\mvn_repo\.m2\repository\org\apache\tomcat\embed\tomcat-embed-core\9.0.52\tomcat-embed-core-9.0.52.jar;D:\software_cluster\mvn_repo\.m2\repository\org\apache\tomcat\embed\tomcat-embed-el\9.0.52\tomcat-embed-el-9.0.52.jar;D:\software_cluster\mvn_repo\.m2\repository\org\apache\tomcat\embed\tomcat-embed-websocket\9.0.52\tomcat-embed-websocket-9.0.52.jar;D:\software_cluster\mvn_repo\.m2\repository\org\springframework\spring-web\5.3.9\spring-web-5.3.9.jar;D:\software_cluster\mvn_repo\.m2\repository\org\springframework\spring-beans\5.3.9\spring-beans-5.3.9.jar;D:\software_cluster\mvn_repo\.m2\repository\org\springframework\spring-webmvc\5.3.9\spring-webmvc-5.3.9.jar;D:\software_cluster\mvn_repo\.m2\repository\org\springframework\spring-aop\5.3.9\spring-aop-5.3.9.jar;D:\software_cluster\mvn_repo\.m2\repository\org\springframework\spring-context\5.3.9\spring-context-5.3.9.jar;D:\software_cluster\mvn_repo\.m2\repository\org\springframework\spring-expression\5.3.9\spring-expression-5.3.9.jar;D:\software_cluster\mvn_repo\.m2\repository\org\apache\kafka\kafka-clients\3.0.0\kafka-clients-3.0.0.jar;D:\software_cluster\mvn_repo\.m2\repository\com\github\luben\zstd-jni\1.5.0-2\zstd-jni-1.5.0-2.jar;D:\software_cluster\mvn_repo\.m2\repository\org\lz4\lz4-java\1.7.1\lz4-java-1.7.1.jar;D:\software_cluster\mvn_repo\.m2\repository\org\xerial\snappy\snappy-java\1.1.8.1\snappy-java-1.1.8.1.jar;D:\software_cluster\mvn_repo\.m2\repository\org\slf4j\slf4j-api\1.7.32\slf4j-api-1.7.32.jar;D:\software_cluster\mvn_repo\.m2\repository\org\slf4j\slf4j-simple\1.7.25\slf4j-simple-1.7.25.jar kafka.consumer.MyConsumer2

这篇关于kafka消费者接收分区测试的文章就介绍到这儿,希望我们推荐的文章对大家有所帮助,也希望大家多多支持为之网!