devbox
Monodyee
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Kafka整合java代码实现生产者与消费者_kafka消费者代码

作者：Monodyee | 2024-05-28 19:11:05

赞

踩

kafka消费者代码

1.引入pom依赖，Kafka和阿里的JSON

  1. 		<dependency>
  2. 			<groupId>org.apache.kafka</groupId>
  3. 			<artifactId>kafka-clients</artifactId>
  4. 			<version>2.8.0</version>
  5. 		</dependency>
  6.  
  7. 		<dependency>
  8. 			<groupId>com.alibaba</groupId>
  9. 			<artifactId>fastjson</artifactId>
  10. 			<version>1.2.41</version>
  11. 		</dependency>

2.实现kafka的生产者

2.1异步发送kafka生产者代码实现

  1. package com.ztesoft.kafka;
  2.  
  3. import java.util.HashMap;
  4. import java.util.Map;
  5. import java.util.Properties;
  6.  
  7. import org.apache.kafka.clients.producer.Callback;
  8. import org.apache.kafka.clients.producer.KafkaProducer;
  9. import org.apache.kafka.clients.producer.Producer;
  10. import org.apache.kafka.clients.producer.ProducerConfig;
  11. import org.apache.kafka.clients.producer.ProducerRecord;
  12. import org.apache.kafka.clients.producer.RecordMetadata;
  13.  
  14. import com.alibaba.fastjson.JSON;
  15.  
  16. /**
  17.  * kafka生产者代码
  18.  *
  19.  */
  20. public class MyKafkaProducer {
  21.  
  22. 	public static void main(String[] args) throws InterruptedException {
  23. 		Properties props = new Properties();
  24. 		// kafka集群，broker-list
  25. 		props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "172.21.72.166:9092");
  26. 		props.put(ProducerConfig.ACKS_CONFIG, "all");
  27. 		// 重试次数
  28. 		props.put("retries", 1);
  29. 		// 批次大小
  30. 		props.put("batch.size", 16384);
  31. 		// 等待时间
  32. 		props.put("linger.ms", 1);
  33. 		// RecordAccumulator缓冲区大小
  34. 		props.put("buffer.memory", 33554432);
  35. 		// 设置序列化
  36. 		props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
  37. 		props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
  38.  
  39. 		Producer<String, String> producer = new KafkaProducer<String, String>(props);
  40. 		for (int i = 100; i < 200; i++) {
  41. 			Map<String, Object> recordMap = new HashMap<String, Object>(20);
  42. 			recordMap.put("seq", i);
  43. 			recordMap.put("name", "测试" + i);
  44. 			recordMap.put("age", i % 20);
  45. 			ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>("user_topic",
  46. 					String.valueOf(i), JSON.toJSONString(recordMap));
  47. 			producer.send(producerRecord, new Callback() {
  48. 				// 回调函数，该方法会在Producer收到ack时调用，为异步调用
  49. 				public void onCompletion(RecordMetadata metadata, Exception e) {
  50. 					String topic = metadata.topic();
  51. 					int partition = metadata.partition();
  52. 					long offset = metadata.offset();
  53. 					if (e != null) {
  54. 						System.out.printf("消息发送失败:topic=%s,offset=%d,key=%d,error=%s%n", topic, partition, offset,
  55. 								e.getMessage());
  56.  
  57. 					} else {
  58. 						System.out.printf("消息发送成功:topic=%s,offset=%d,key=%d,value=%s%n", topic, partition, offset);
  59. 					}
  60. 				}
  61. 			});
  62.  
  63. 			Thread.sleep(2 * 1000);
  64.  
  65. 		}
  66. 		producer.close();
  67. 	}
  68. }

2.2同步发送kafka生产者代码实现

  1. package com.ztesoft.kafka;
  2.  
  3. import java.util.HashMap;
  4. import java.util.Map;
  5. import java.util.Properties;
  6. import java.util.concurrent.ExecutionException;
  7.  
  8. import org.apache.kafka.clients.producer.Callback;
  9. import org.apache.kafka.clients.producer.KafkaProducer;
  10. import org.apache.kafka.clients.producer.Producer;
  11. import org.apache.kafka.clients.producer.ProducerConfig;
  12. import org.apache.kafka.clients.producer.ProducerRecord;
  13. import org.apache.kafka.clients.producer.RecordMetadata;
  14.  
  15. import com.alibaba.fastjson.JSON;
  16.  
  17. /**
  18.  * kafka生产者代码
  19.  *
  20.  */
  21. public class MyKafkaProducer {
  22.  
  23. 	public static void main(String[] args) throws InterruptedException, ExecutionException {
  24. 		Properties props = new Properties();
  25. 		// kafka集群，broker-list
  26. 		props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "172.21.72.166:9092");
  27. 		props.put(ProducerConfig.ACKS_CONFIG, "all");
  28. 		// 重试次数
  29. 		props.put("retries", 1);
  30. 		// 批次大小
  31. 		props.put("batch.size", 16384);
  32. 		// 等待时间
  33. 		props.put("linger.ms", 1);
  34. 		// RecordAccumulator缓冲区大小
  35. 		props.put("buffer.memory", 33554432);
  36. 		// 设置序列化
  37. 		props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
  38. 		props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
  39.  
  40. 		Producer<String, String> producer = new KafkaProducer<String, String>(props);
  41. 		for (int i = 100; i < 200; i++) {
  42. 			Map<String, Object> recordMap = new HashMap<String, Object>(20);
  43. 			recordMap.put("seq", i);
  44. 			recordMap.put("name", "测试" + i);
  45. 			recordMap.put("age", i % 20);
  46. 			ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>("user_topic",
  47. 					String.valueOf(i), JSON.toJSONString(recordMap));
  48. 			//调用get方法为同步发送返回
  49. 			producer.send(producerRecord).get();
  50. 			Thread.sleep(2 * 1000);
  51.  
  52. 		}
  53. 		producer.close();
  54. 	}
  55. }

3.实现Kafka的消费者

3.1实现kafka消费的自动提交offset

  1. package com.ztesoft.kafka;
  2.  
  3. import java.util.Arrays;
  4. import java.util.Properties;
  5.  
  6. import org.apache.kafka.clients.consumer.ConsumerRecord;
  7. import org.apache.kafka.clients.consumer.ConsumerRecords;
  8. import org.apache.kafka.clients.consumer.KafkaConsumer;
  9.  
  10.  
  11.  
  12. public class MyKafkaConsumer {
  13.  
  14. 	public static void main(String[] args) {
  15.  
  16. 		Properties props = new Properties();
  17. 		// kafka地址
  18. 		props.put("bootstrap.servers", "172.21.72.166:9092");
  19. 		// 设置消费组
  20. 		props.put("group.id", "bigdata");
  21. 		// 是否自动提交
  22. 		props.put("enable.auto.commit", "true");
  23. 		// 设置自动提交时间隔
  24. 		props.put("auto.commit.interval.ms", "1000");
  25. 		// 设置消费,一般设置earliest或者latest
  26. 		props.put("auto.offset.reset", "earliest");
  27. 		// 序列化
  28. 		props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  29. 		props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  30. 		KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
  31. 		consumer.subscribe(Arrays.asList("user_topic"));
  32. 		while (true) {
  33. 			ConsumerRecords<String, String> records = consumer.poll(10);
  34. 			for (ConsumerRecord<String, String> record : records) {
  35. 				System.out.printf("partition = %d,offset = %d, key = %s, value = %s%n", record.partition(),
  36. 						record.offset(), record.key(), record.value());
  37. 				int partition = record.partition();
  38. 				long offset = record.offset();
  39. 				String key = record.key();
  40. 				String value = record.value();
  41. 				// 打印消费参数
  42. 				System.out.printf("partition = %d,offset = %d, key = %s, value = %s%n", partition, offset, key, value);
  43.  
  44. 			}
  45. 			consumer.commitAsync();
  46. 		}
  47. 	}
  48.  
  49. }

3.2实现kafka消费的手动提交offset

  1. package com.ztesoft.kafka;
  2.  
  3. import java.util.Arrays;
  4. import java.util.Properties;
  5.  
  6. import org.apache.kafka.clients.consumer.ConsumerRecord;
  7. import org.apache.kafka.clients.consumer.ConsumerRecords;
  8. import org.apache.kafka.clients.consumer.KafkaConsumer;
  9.  
  10.  
  11.  
  12. public class MyKafkaConsumer {
  13.  
  14. 	public static void main(String[] args) {
  15.  
  16. 		Properties props = new Properties();
  17. 		// kafka地址
  18. 		props.put("bootstrap.servers", "172.21.72.166:9092");
  19. 		// 设置消费组
  20. 		props.put("group.id", "bigdata");
  21. 		// 是否自动提交
  22. 		props.put("enable.auto.commit", "false");
  23. 		// 设置自动提交时间隔
  24. 		props.put("auto.commit.interval.ms", "1000");
  25. 		// 设置消费,一般设置earliest或者latest
  26. 		props.put("auto.offset.reset", "earliest");
  27. 		// 序列化
  28. 		props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  29. 		props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  30. 		KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
  31. 		consumer.subscribe(Arrays.asList("user_topic"));
  32. 		while (true) {
  33. 			ConsumerRecords<String, String> records = consumer.poll(10);
  34. 			for (ConsumerRecord<String, String> record : records) {
  35. 				System.out.printf("partition = %d,offset = %d, key = %s, value = %s%n", record.partition(),
  36. 						record.offset(), record.key(), record.value());
  37. 				int partition = record.partition();
  38. 				long offset = record.offset();
  39. 				String key = record.key();
  40. 				String value = record.value();
  41. 				// 打印消费参数
  42. 				System.out.printf("partition = %d,offset = %d, key = %s, value = %s%n", partition, offset, key, value);
  43.  
  44. 			}
  45. 			//每次消费完数据进行异步提交
  46. 			consumer.commitAsync();
  47. 		}
  48. 	}
  49.  
  50. }

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/Monodyee/article/detail/638904
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号