• 3、Kafka进阶提升-消费者

     

     

     

     消费者与消费者组代码：

    1. import com.lvxiaosha.kafka.api.Const;
    2. import org.apache.kafka.clients.consumer.ConsumerConfig;
    3. import org.apache.kafka.clients.consumer.ConsumerRecord;
    4. import org.apache.kafka.clients.consumer.ConsumerRecords;
    5. import org.apache.kafka.clients.consumer.KafkaConsumer;
    6. import org.apache.kafka.common.TopicPartition;
    7.  
    8. import java.time.Duration;
    9. import java.util.Collections;
    10. import java.util.List;
    11. import java.util.Properties;
    12.  
    13. public class Consumer1 {
    14.  
    15. 	public static void main(String[] args) {
    16. 		Properties props = new Properties();
    17. 		props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.110.130:9092");
    18. 		//	GROUP_ID_CONFIG 消费者组配置
    19. 		props.put(ConsumerConfig.GROUP_ID_CONFIG, "module-group-id-1"); 
    20. 		props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true); 
    21. 		props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000"); 
    22. 		props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer"); 
    23. 		props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer"); 
    24. 		KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props); 
    25. 		//	订阅主题
    26. 		consumer.subscribe(Collections.singletonList(Const.TOPIC_MODULE));
    27. 		System.err.println("consumer1 started.. ");
    28. 		try {
    29. 	        while (true) {
    30. 	            // 	拉取结果集
    31. 	        	ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    32. 	            for (TopicPartition partition : records.partitions()) {
    33. 	                List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
    34. 	                String topic = partition.topic();
    35. 	                int size = partitionRecords.size();
    36. 	                System.err.println(String.format("---- 获取topic: %s, 分区位置：%s, 消息数为：%s ----",topic, partition.partition(), size));
    37. 	                for (int i = 0; i< size; i++) {
    38. 	                    long offset = partitionRecords.get(i).offset() + 1;
    39. 		                System.err.println(String.format("获取value: %s, 提交的 offset: %s", 
    40. 		                		partitionRecords.get(i).value(), offset)); 
    41. 	                }
    42. 	            }
    43. 	        }			
    44. 		} finally {
    45. 			consumer.close();
    46. 		}
    47. 	}
    48. }
    1. import com.lvxiaosha.kafka.api.Const;
    2. import org.apache.kafka.clients.consumer.ConsumerConfig;
    3. import org.apache.kafka.clients.consumer.ConsumerRecord;
    4. import org.apache.kafka.clients.consumer.ConsumerRecords;
    5. import org.apache.kafka.clients.consumer.KafkaConsumer;
    6. import org.apache.kafka.common.TopicPartition;
    7.  
    8. import java.time.Duration;
    9. import java.util.Collections;
    10. import java.util.List;
    11. import java.util.Properties;
    12.  
    13. public class Consumer2 {
    14.  
    15. 	public static void main(String[] args) {
    16. 		Properties props = new Properties();
    17. 		props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.110.130:9092");
    18. 		props.put(ConsumerConfig.GROUP_ID_CONFIG, "module-group-id-2"); 
    19. 		props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true); 
    20. 		props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000"); 
    21. 		props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer"); 
    22. 		props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer"); 
    23. 		KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props); 
    24. 		//	订阅主题
    25. 		consumer.subscribe(Collections.singletonList(Const.TOPIC_MODULE));
    26. 		System.err.println("consumer2 started.. ");
    27. 		try {
    28. 	        while (true) {
    29. 	            // 	拉取结果集
    30. 	        	ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    31. 	            for (TopicPartition partition : records.partitions()) {
    32. 	                List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
    33. 	                String topic = partition.topic();
    34. 	                int size = partitionRecords.size();
    35. 	                System.err.println(String.format("---- 获取topic: %s, 分区位置：%s, 消息数为：%s ----",topic, partition.partition(), size));
    36. 	                for (int i = 0; i< size; i++) {
    37. 	                    long offset = partitionRecords.get(i).offset() + 1;
    38. 		                System.err.println(String.format("获取value: %s, 提交的 offset: %s", 
    39. 		                		partitionRecords.get(i).value(), offset)); 
    40. 	                }
    41. 	            }
    42. 	        }			
    43. 		} finally {
    44. 			consumer.close();
    45. 		}
    46. 	}
    47. }
    1. import com.alibaba.fastjson.JSON;
    2. import com.lvxiaosha.kafka.api.Const;
    3. import com.lvxiaosha.kafka.api.User;
    4. import org.apache.kafka.clients.producer.KafkaProducer;
    5. import org.apache.kafka.clients.producer.ProducerConfig;
    6. import org.apache.kafka.clients.producer.ProducerRecord;
    7.  
    8. import java.util.Properties;
    9.  
    10. public class Producer {
    11.  
    12. 	public static void main(String[] args) {
    13. 		
    14. 		Properties props = new Properties(); 
    15. 		props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.110.130:9092");
    16. 		props.put(ProducerConfig.CLIENT_ID_CONFIG, "module-producer"); 
    17. 		props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); 
    18. 		props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
    19. 		KafkaProducer<String, String> producer = new KafkaProducer<>(props);	
    20. 		
    21. 		for(int i = 0 ; i < 10; i ++) {
    22. 			User user = new User();
    23. 			user.setId(i+"");
    24. 			user.setName("张三");
    25. 			producer.send(new ProducerRecord<>(Const.TOPIC_MODULE, JSON.toJSONString(user)));	
    26. 		}
    27. 		
    28. 		producer.close();
    29. 		
    30. 	}
    31. 	
    32. }

    注：如果Consumer1和Consumer2里面消费者的消费者组配置(props.put(ConsumerConfig.GROUP_ID_CONFIG, "module-group-id-1");)都是一样的话，说明两个消费者隶属于同一个消费者组，相当于点对点模型，Consumer1和Consumer2这时会均摊的消费Topic的消息。反之，如果Consumer1和Consumer2里面消费者的消费者组配置不一样，相当于发布订阅模型，Consumer1和Consumer2这时会同时消费Topic里面的消息。

    Kafka消费者必要参数方法(没有这些参数就启动不起来)

    • bootstrap.servers:用来指定连接Kafka集群所需的broker地址清单。
    • key.deserializer和value.deserizer:反序列化参数；
    • group.id:消费者所属消费组；
    • subscribe:消费主题订阅，支持集合/标准正则表达式；
    • assign:只订阅主题的某个分区。

    Kafka消费者提交位移

     

    主题订阅代码：

    1. import com.lvxiaosha.kafka.api.Const;
    2. import org.apache.kafka.clients.consumer.ConsumerConfig;
    3. import org.apache.kafka.clients.consumer.ConsumerRecord;
    4. import org.apache.kafka.clients.consumer.ConsumerRecords;
    5. import org.apache.kafka.clients.consumer.KafkaConsumer;
    6. import org.apache.kafka.common.PartitionInfo;
    7. import org.apache.kafka.common.TopicPartition;
    8. import org.apache.kafka.common.serialization.StringDeserializer;
    9.  
    10. import java.time.Duration;
    11. import java.util.ArrayList;
    12. import java.util.List;
    13. import java.util.Properties;
    14.  
    15. public class CoreConsumer {
    16.  
    17. 	public static void main(String[] args) {
    18. 		Properties properties = new Properties();
    19. 		properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.110.130:9092");
    20. 		properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    21. 		properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    22. 		properties.put(ConsumerConfig.GROUP_ID_CONFIG, "core-group");
    23. 		
    24. 		//	TODO : 使用手工方式提交
    25. 		properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
    26. 		properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 5000);
    27. 		KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
    28. 		
    29. 		//	对于consume消息的订阅 subscribe方法 ：可以订阅一个 或者  多个 topic
    30. //		consumer.subscribe(Collections.singletonList(Const.TOPIC_CORE));
    31. 		//	也可以支持正则表达式方式的订阅 
    32. //		consumer.subscribe(Pattern.compile("topic-.*"));
    33. 		
    34. 		//	可以指定订阅某个主题下的某一个 或者多个 partition
    35. //		consumer.assign(Arrays.asList(new TopicPartition(Const.TOPIC_CORE, 0), new TopicPartition(Const.TOPIC_CORE, 2)));
    36.  
    37. 		//	如何拉取主题下的所有partition
    38. 		List<TopicPartition> tpList = new ArrayList<TopicPartition>();
    39. 		List<PartitionInfo> tpinfoList = consumer.partitionsFor(Const.TOPIC_CORE);
    40. 		for(PartitionInfo pi : tpinfoList) {
    41. 			System.err.println("主题:"+ pi.topic() +", 分区: " + pi.partition());
    42. 			tpList.add(new TopicPartition(pi.topic(), pi.partition()));
    43. 		}
    44. 		
    45. 		consumer.assign(tpList);
    46. 		
    47. 		System.err.println("core consumer started...");
    48. 		
    49. 		try {
    50. 			while(true) {
    51. 				ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    52. 				for(TopicPartition topicPartition : records.partitions()) {
    53. 					List<ConsumerRecord<String, String>> partitionRecords = records.records(topicPartition);
    54. 					String topic = topicPartition.topic();
    55. 					int size = partitionRecords.size();
    56. 					System.err.println(String.format("--- 获取topic: %s, 分区位置：%s, 消息总数： %s", topic, topicPartition.partition(), size));
    57. 					for(int i = 0; i < size; i++) {
    58. 						ConsumerRecord<String, String> consumerRecord = partitionRecords.get(i);
    59. 						String value = consumerRecord.value();
    60. 						long offset = consumerRecord.offset();
    61. 						long commitOffser = offset + 1;
    62. 						System.err.println(String.format("获取实际消息 value：%s, 消息offset: %s, 提交offset: %s", value, offset, commitOffser));
    63. 					}
    64. 				}
    65. 			}			
    66. 		} finally {
    67. 			consumer.close();
    68. 		}
    69. 	}
    70. }
    1. import com.alibaba.fastjson.JSON;
    2. import com.lvxiaosha.kafka.api.Const;
    3. import com.lvxiaosha.kafka.api.User;
    4. import org.apache.kafka.clients.producer.KafkaProducer;
    5. import org.apache.kafka.clients.producer.ProducerConfig;
    6. import org.apache.kafka.clients.producer.ProducerRecord;
    7. import org.apache.kafka.common.serialization.StringSerializer;
    8.  
    9. import java.util.Properties;
    10.  
    11. public class CoreProducer {
    12.  
    13. 	public static void main(String[] args) {
    14. 		Properties properties = new Properties();
    15. 		properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.110.130:9092");
    16. 		properties.put(ProducerConfig.CLIENT_ID_CONFIG, "core-producer");
    17. 		properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    18. 		properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    19. 		
    20. 		KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
    21. 		for(int i = 0; i <10; i ++) {
    22. 			User user = new User("00" + i, "张三");
    23. 			ProducerRecord<String, String> record = 
    24. 					new ProducerRecord<String, String>(Const.TOPIC_CORE,
    25. 							JSON.toJSONString(user));
    26. 			producer.send(record);			
    27. 		}
    28. 		producer.close();
    29. 	}
    30. }

    手工提交代码：

    1. import com.lvxiaosha.kafka.api.Const;
    2. import org.apache.kafka.clients.consumer.*;
    3. import org.apache.kafka.common.TopicPartition;
    4. import org.apache.kafka.common.serialization.StringDeserializer;
    5.  
    6. import java.time.Duration;
    7. import java.util.Collections;
    8. import java.util.List;
    9. import java.util.Map;
    10. import java.util.Properties;
    11.  
    12. public class CommitConsumer {
    13.  
    14. 	public static void main(String[] args) {
    15. 		Properties properties = new Properties();
    16. 		properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.110.130:9092");
    17. 		properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    18. 		properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    19. 		properties.put(ConsumerConfig.GROUP_ID_CONFIG, "core-group");
    20. 		
    21. 		//	使用手工方式提交
    22. 		properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
    23. 		
    24. //		properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
    25. //		properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 5000);
    26. 		
    27. 		//	消费者默认每次拉取的位置：从什么位置开始拉取消息
    28. 		//	AUTO_OFFSET_RESET_CONFIG 有三种方式： "latest", "earliest", "none"
    29. 		//	none
    30. 		//	latest	从一个分区的最后提交的offset开始拉取消息
    31. 		//	earliset	从最开始的起始位置拉取消息 0
    32. 		properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "");
    33. 		
    34. 		
    35. 		KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
    36. 		
    37. 		//	对于consume消息的订阅 subscribe方法 ：可以订阅一个 或者  多个 topic
    38. 		consumer.subscribe(Collections.singletonList(Const.TOPIC_CORE));
    39. 		
    40. 		System.err.println("core consumer started...");
    41. 		
    42. 		try {
    43. 			while(true) {
    44. 				ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    45. 				for(TopicPartition topicPartition : records.partitions()) {
    46. 					List<ConsumerRecord<String, String>> partitionRecords = records.records(topicPartition);
    47. 					String topic = topicPartition.topic();
    48. 					int size = partitionRecords.size();
    49. 					System.err.println(String.format("--- 获取topic: %s, 分区位置：%s, 消息总数： %s", topic, topicPartition.partition(), size));
    50. 					for(int i = 0; i < size; i++) {
    51. 						ConsumerRecord<String, String> consumerRecord = partitionRecords.get(i);
    52. 						String value = consumerRecord.value();
    53. 						long offset = consumerRecord.offset();
    54. 						long commitOffser = offset + 1;
    55. 						System.err.println(String.format("获取实际消息 value：%s, 消息offset: %s, 提交offset: %s", value, offset, commitOffser));
    56. 						//	在一个partition内部，每一条消息记录 进行一一提交方式
    57. //						consumer.commitSync(Collections.singletonMap(topicPartition, new OffsetAndMetadata(commitOffser)));
    58. 						consumer.commitAsync(
    59. 								Collections.singletonMap(topicPartition, new OffsetAndMetadata(commitOffser)),
    60. 								new OffsetCommitCallback() {
    61. 									@Override
    62. 									public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets,
    63. 											Exception exception) {
    64. 										if (null != exception) {
    65. 											System.err.println("error . 处理");
    66. 										}
    67. 										System.err.println("异步提交成功：" + offsets);
    68. 									}
    69. 								});
    70. 					
    71. 					}
    72. 					//	一个partition做一次提交动作
    73. 				}
    74. 				
    75. 				/**
    76. 				//	整体提交：同步方式
    77. //				consumer.commitSync();
    78. 				
    79. 				//	整体提交：异步方式
    80. 				consumer.commitAsync(new OffsetCommitCallback() {
    81. 					@Override
    82. 					public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
    83. 						if(null != exception) {
    84. 							System.err.println("error . 处理");
    85. 						}
    86. 						System.err.println("异步提交成功：" + offsets);
    87. 					}
    88. 				});
    89. 				*/
    90. 				
    91. 			}			
    92. 		} finally {
    93. 			consumer.close();
    94. 		}
    95. 	}
    96. }
    1. import com.alibaba.fastjson.JSON;
    2. import com.lvxiaosha.kafka.api.Const;
    3. import com.lvxiaosha.kafka.api.User;
    4. import org.apache.kafka.clients.producer.KafkaProducer;
    5. import org.apache.kafka.clients.producer.ProducerConfig;
    6. import org.apache.kafka.clients.producer.ProducerRecord;
    7. import org.apache.kafka.common.serialization.StringSerializer;
    8.  
    9. import java.util.Properties;
    10.  
    11. public class CommitProducer {
    12.  
    13. 	public static void main(String[] args) {
    14. 		Properties properties = new Properties();
    15. 		properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.110.130:9092");
    16. 		properties.put(ProducerConfig.CLIENT_ID_CONFIG, "core-producer");
    17. 		properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    18. 		properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    19. 		
    20. 		KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
    21. 		for(int i = 0; i <10; i ++) {
    22. 			User user = new User("00" + i, "张三");
    23. 			ProducerRecord<String, String> record = 
    24. 					new ProducerRecord<String, String>(Const.TOPIC_CORE,
    25. 							JSON.toJSONString(user));
    26. 			producer.send(record);			
    27. 		}
    28. 		producer.close();
    29. 	}
    30. }

    谈一谈 Kafka 的再均衡？

    在Kafka中，当有新消费者加入或者订阅的topic数发生变化时，会触发Rebalance(再均衡：在同一个消费者组当中，分区的所有权从一个消费者转移到另外一个消费者)机制，Rebalance顾名思义就是重新均衡消费者消费。Rebalance的过程如下：

    第一步：所有成员都向coordinator发送请求，请求入组。一旦所有成员都发送了请求，coordinator会从中选择一个consumer担任leader的角色，并把组成员信息以及订阅信息发给leader。第二步：leader开始分配消费方案，指明具体哪个consumer负责消费哪些topic的哪些partition。一旦完成分配，leader会将这个方案发给coordinator。coordinator接收到分配方案之后会把方案发给各个consumer，这样组内的所有成员就都知道自己应该消费哪些分区了。所以对于Rebalance来说，Coordinator起着至关重要的作用。

     Kafka再均衡监听代码：

    1. import com.lvxiaosha.kafka.api.Const;
    2. import org.apache.kafka.clients.consumer.*;
    3. import org.apache.kafka.common.TopicPartition;
    4.  
    5. import java.time.Duration;
    6. import java.util.Collection;
    7. import java.util.Collections;
    8. import java.util.List;
    9. import java.util.Properties;
    10.  
    11. public class RebalanceConsumer1 {
    12.  
    13. 	public static void main(String[] args) {
    14. 		Properties props = new Properties();
    15. 		props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.110.130:9092");
    16. 		//	GROUP_ID_CONFIG 消费者组配置
    17. 		props.put(ConsumerConfig.GROUP_ID_CONFIG, "rebalance-group"); 
    18. 		props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true); 
    19. 		props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000"); 
    20. 		props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer"); 
    21. 		props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer"); 
    22. 		KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props); 
    23. 		//	订阅主题
    24. 		consumer.subscribe(Collections.singletonList(Const.TOPIC_REBALANCE), new ConsumerRebalanceListener() {
    25. 			@Override
    26. 			public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
    27. 				System.err.println("Revoked Partitions:" + partitions);
    28. 			}
    29. 			
    30. 			@Override
    31. 			public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
    32. 				System.err.println("AssignedAssigned Partitions:" + partitions);
    33. 			}
    34. 		});
    35. 		System.err.println("rebalance consumer1 started.. ");
    36. 		try {
    37. 	        while (true) {
    38. 	            // 	拉取结果集
    39. 	        	ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    40. 	            for (TopicPartition partition : records.partitions()) {
    41. 	                List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
    42. 	                String topic = partition.topic();
    43. 	                int size = partitionRecords.size();
    44. 	                System.err.println(String.format("---- 获取topic: %s, 分区位置：%s, 消息数为：%s ----",topic, partition.partition(), size));
    45. 	                for (int i = 0; i< size; i++) {
    46. 	                    long offset = partitionRecords.get(i).offset() + 1;
    47. 		                System.err.println(String.format("获取value: %s, 提交的 offset: %s", 
    48. 		                		partitionRecords.get(i).value(), offset)); 
    49. 	                }
    50. 	            }
    51. 	        }			
    52. 		} finally {
    53. 			consumer.close();
    54. 		}
    55. 	}
    56. }
    1. import com.lvxiaosha.kafka.api.Const;
    2. import org.apache.kafka.clients.consumer.*;
    3. import org.apache.kafka.common.TopicPartition;
    4.  
    5. import java.time.Duration;
    6. import java.util.Collection;
    7. import java.util.Collections;
    8. import java.util.List;
    9. import java.util.Properties;
    10.  
    11. public class RebalanceConsumer2 {
    12.  
    13. 	public static void main(String[] args) {
    14. 		Properties props = new Properties();
    15. 		props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.110.130:9092");
    16. 		props.put(ConsumerConfig.GROUP_ID_CONFIG, "rebalance-group"); 
    17. 		props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true); 
    18. 		props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000"); 
    19. 		props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer"); 
    20. 		props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer"); 
    21. 		KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props); 
    22. 		//	订阅主题
    23. 		//	订阅主题
    24. 		consumer.subscribe(Collections.singletonList(Const.TOPIC_REBALANCE), new ConsumerRebalanceListener() {
    25. 			@Override
    26. 			public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
    27. 				System.err.println("Revoked Partitions:" + partitions);
    28. 			}
    29. 			
    30. 			@Override
    31. 			public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
    32. 				System.err.println("AssignedAssigned Partitions:" + partitions);
    33. 			}
    34. 		});
    35. 		
    36. 		System.err.println("rebalance consumer2 started.. ");
    37. 		try {
    38. 	        while (true) {
    39. 	            // 	拉取结果集
    40. 	        	ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    41. 	            for (TopicPartition partition : records.partitions()) {
    42. 	                List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
    43. 	                String topic = partition.topic();
    44. 	                int size = partitionRecords.size();
    45. 	                System.err.println(String.format("---- 获取topic: %s, 分区位置：%s, 消息数为：%s ----",topic, partition.partition(), size));
    46. 	                for (int i = 0; i< size; i++) {
    47. 	                    long offset = partitionRecords.get(i).offset() + 1;
    48. 		                System.err.println(String.format("获取value: %s, 提交的 offset: %s", 
    49. 		                		partitionRecords.get(i).value(), offset)); 
    50. 	                }
    51. 	            }
    52. 	        }			
    53. 		} finally {
    54. 			consumer.close();
    55. 		}
    56. 	}
    57. }
    1. import com.alibaba.fastjson.JSON;
    2. import com.lvxiaosha.kafka.api.Const;
    3. import com.lvxiaosha.kafka.api.User;
    4. import org.apache.kafka.clients.producer.KafkaProducer;
    5. import org.apache.kafka.clients.producer.ProducerConfig;
    6. import org.apache.kafka.clients.producer.ProducerRecord;
    7.  
    8. import java.util.Properties;
    9.  
    10. public class RebalanceProducer {
    11.  
    12. 	public static void main(String[] args) {
    13. 		
    14. 		Properties props = new Properties(); 
    15. 		props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.110.130:9092");
    16. 		props.put(ProducerConfig.CLIENT_ID_CONFIG, "rebalance-producer"); 
    17. 		props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); 
    18. 		props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
    19. 		KafkaProducer<String, String> producer = new KafkaProducer<>(props);	
    20. 		
    21. 		for(int i = 0 ; i < 10; i ++) {
    22. 			User user = new User();
    23. 			user.setId(i+"");
    24. 			user.setName("张三");
    25. 			producer.send(new ProducerRecord<>(Const.TOPIC_REBALANCE, JSON.toJSONString(user)));	
    26. 		}
    27. 		
    28. 		producer.close();
    29. 		
    30. 	}
    31. 	
    32. }

    Kafka消费者多线程

     

     消费者多线程模型1代码实现：

    1. import org.apache.kafka.clients.consumer.ConsumerRecord;
    2. import org.apache.kafka.clients.consumer.ConsumerRecords;
    3. import org.apache.kafka.clients.consumer.KafkaConsumer;
    4. import org.apache.kafka.common.TopicPartition;
    5.  
    6. import java.time.Duration;
    7. import java.util.Arrays;
    8. import java.util.List;
    9. import java.util.Properties;
    10. import java.util.concurrent.atomic.AtomicInteger;
    11.  
    12. public class KafkaConsumerMt1 implements Runnable {
    13.  
    14. 	private KafkaConsumer<String, String> consumer;
    15. 	
    16. 	private volatile boolean isRunning = true;
    17. 	
    18. 	private static AtomicInteger counter = new AtomicInteger(0);
    19. 	
    20. 	private String consumerName;
    21. 	
    22. 	public KafkaConsumerMt1(Properties properties, String topic) {
    23. 		this.consumer = new KafkaConsumer<>(properties);
    24. 		this.consumer.subscribe(Arrays.asList(topic));
    25. 		this.consumerName = "KafkaConsumerMt1-" + counter.getAndIncrement();
    26. 		System.err.println(this.consumerName + " started ");
    27. 	}
    28. 	
    29. 	@Override
    30. 	public void run() {
    31. 		try {
    32. 			while(isRunning) {
    33. 				//	包含所有topic下的所有消息内容
    34. 				ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofMillis(1000));
    35. 				for(TopicPartition topicPartition : consumerRecords.partitions()) {
    36. //					String topic = topicPartition.topic();
    37. 					//	根据具体的topicPartition 去获取对应topicPartition下的数据集合
    38. 					List<ConsumerRecord<String, String>> partitionList = consumerRecords.records(topicPartition);
    39. 					int size = partitionList.size();
    40. 					for(int i = 0; i < size; i++) {
    41. 						ConsumerRecord<String, String> consumerRecord = partitionList.get(i);
    42. 						// do execute message
    43. 						String message = consumerRecord.value();
    44. 						long messageOffset = consumerRecord.offset();
    45. 						System.err.println("当前消费者："+ consumerName 
    46. 								+ ",消息内容：" + message 
    47. 								+ ", 消息的偏移量: " + messageOffset
    48. 								+ "当前线程：" + Thread.currentThread().getName());
    49. 					}
    50. 				}
    51. 			}
    52. 		} finally {
    53. 			if(consumer != null) {
    54. 				consumer.close();
    55. 			}
    56. 		}
    57. 	}
    58.  
    59. 	public boolean isRunning() {
    60. 		return isRunning;
    61. 	}
    62.  
    63. 	public void setRunning(boolean isRunning) {
    64. 		this.isRunning = isRunning;
    65. 	}
    66.  
    67. }
    1. import com.alibaba.fastjson.JSON;
    2. import com.lvxiaosha.kafka.api.Const;
    3. import com.lvxiaosha.kafka.api.User;
    4. import org.apache.kafka.clients.producer.KafkaProducer;
    5. import org.apache.kafka.clients.producer.ProducerConfig;
    6. import org.apache.kafka.clients.producer.ProducerRecord;
    7.  
    8. import java.util.Properties;
    9. /**
    10.  * $Mt1Producer
    11.  * @author hezhuo.bai
    12.  * @since 2019年2月28日 下午12:38:46
    13.  */
    14. public class Mt1Producer {
    15.  
    16. 	public static void main(String[] args) {
    17. 		
    18. 		Properties props = new Properties(); 
    19. 		props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.110.130:9092");
    20. 		props.put(ProducerConfig.CLIENT_ID_CONFIG, "mt1-producer"); 
    21. 		props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); 
    22. 		props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
    23. 		KafkaProducer<String, String> producer = new KafkaProducer<>(props);	
    24. 		
    25. 		for(int i = 0 ; i < 10; i ++) {
    26. 			User user = new User();
    27. 			user.setId(i+"");
    28. 			user.setName("张三");
    29. 			producer.send(new ProducerRecord<>(Const.TOPIC_MT1, JSON.toJSONString(user)));	
    30. 		}
    31. 		
    32. 		producer.close();
    33. 		
    34. 	}
    35.  
    36. 	
    37. }
    1. import com.lvxiaosha.kafka.api.Const;
    2. import org.apache.kafka.clients.consumer.ConsumerConfig;
    3.  
    4. import java.util.Properties;
    5. import java.util.concurrent.ExecutorService;
    6. import java.util.concurrent.Executors;
    7.  
    8. public class Mt1Test {
    9.  
    10. 	public static void main(String[] args) {
    11. 		
    12. 		Properties props = new Properties();
    13. 		props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.110.130:9092");
    14. 		props.put(ConsumerConfig.GROUP_ID_CONFIG, "mt1-group"); 
    15. 		//	自动提交的方式
    16. 		props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true); 
    17. 		props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000"); 
    18. 		props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer"); 
    19. 		props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer"); 
    20. 		String topic = Const.TOPIC_MT1;
    21. 		
    22. 		// coreSize
    23. 		int coreSize = 5;
    24. 		ExecutorService executorService = Executors.newFixedThreadPool(coreSize);
    25. 		
    26. 		for(int i =0; i <5; i++) {
    27. 			executorService.execute(new KafkaConsumerMt1(props, topic));
    28. 		}
    29. 		
    30. 	}
    31. }

    采用Master-worker模型（master负责分发，worker负责处理）。master初始化阶段，创建多个worker实例，然后master负责拉取消息，并负责分发给各个worker执行。各个worker执行完成后，将执行结果回传给master，由master统一进行commit。

    这样避免了多个线程共同操作consumer，导致kafka抛异常ConcurrentModifacationException

    Master-worker模型

    Kafka消费者重要参数

     

  • 相关阅读:
    Vue - 实现点击按钮（笔图标）可编辑 input 输入框（点击文字内容后变成 ＜input＞ 输入框同时能修改和取消、删除）
    vue中的mixin混入
    java毕业设计旧衣物捐赠系统（附源码、数据库）
    VUE快速入门-2
    Flutter 按钮 大集合
    映翰通C++ 一面（技术面、65min、offer）
    深入了解iOS内存(WWDC 2018)笔记-内存诊断
    淘宝价格带卡位公式是什么?如何定价？
    前端必看！css中常用单位及区别
    六、02【Java 多线程】之线程基础知识
  • 原文地址：https://blog.csdn.net/Xx13624558575/article/details/126778621