Java实现Kafka生产者和消费者的示例

Java实现Kafka生产者和消费者的示例

Kafka简介

Kafka是由Apache软件基金会开发的一个开源流处理平台，由Scala和Java编写。Kafka的目标是为处理实时数据提供一个统一、高吞吐、低延迟的平台。

文章持续更新，微信搜索「万猫学社」第一时间阅读，关注后回复「电子书」，免费获取12本Java必读技术书籍。

方式一：kafka-clients

引入依赖

在pom.

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

文章持续更新，微信搜索「万猫学社」第一时间阅读，关注后回复「电子书」，免费获取12本Java必读技术书籍。

生产者

创建一个KafkaProducer的生产者实例：

@Configurationpublic class Config { public final static String bootstrapServers = "127.0.0.1:9092"; @Bean(destroyMethod = "close") public KafkaProducer<String, String> kafkaProducer() {  Properties props = new Properties();  //设置Kafka服务器地址  props.put("bootstrap.servers", bootstrapServers);  //设置数据key的序列化处理类  props.put("key.serializer", StringSerializer.class.getName());  //设置数据value的序列化处理类  props.put("value.serializer", StringSerializer.class.getName());  KafkaProducer<String, String> producer = new KafkaProducer<>(props);  return producer; }}

文章持续更新，微信搜索「万猫学社」第一时间阅读，关注后回复「电子书」，免费获取12本Java必读技术书籍。

在Controller中进行使用：

@RestController@Slf4jpublic class Controller { @Autowired private KafkaProducer<String, String> kafkaProducer; @RequestMapping("/kafkaClientsSend") public String send() {  String uuid = UUID.randomUUID().toString();  RecordMetadata recordMetadata = null;  try {  	//将消息发送到Kafka服务器的名称为"one-more-topic"的Topic中   recordMetadata = kafkaProducer.send(new ProducerRecord<>("one-more-topic", uuid)).get();   log.info("recordMetadata: {}", recordMetadata);   log.info("uuid: {}", uuid);  } catch (Exception e) {   log.error("send fail, uuid: {}", uuid, e);  }  return uuid; }}

文章持续更新，微信搜索「万猫学社」第一时间阅读，关注后回复「电子书」，免费获取12本Java必读技术书籍。

消费者

创建一个KafkaConsumer的消费者实例：

@Configurationpublic class Config { public final static String groupId = "kafka-clients-group"; public final static String bootstrapServers = "127.0.0.1:9092"; @Bean(destroyMethod = "close") public KafkaConsumer<String, String> kafkaConsumer() {  Properties props = new Properties();  //设置Kafka服务器地址  props.put("bootstrap.servers", bootstrapServers);  //设置消费组  props.put("group.id", groupId);  //设置数据key的反序列化处理类  props.put("key.deserializer", StringDeserializer.class.getName());  //设置数据value的反序列化处理类  props.put("value.deserializer", StringDeserializer.class.getName());  props.put("enable.auto.commit", "true");  props.put("auto.commit.interval.ms", "1000");  props.put("session.timeout.ms", "30000");  KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(props);  //订阅名称为"one-more-topic"的Topic的消息  kafkaConsumer.subscribe(Arrays.asList("one-more-topic"));  return kafkaConsumer; }}

文章持续更新，微信搜索「万猫学社」第一时间阅读，关注后回复「电子书」，免费获取12本Java必读技术书籍。

在Controller中进行使用：

@RestController@Slf4jpublic class Controller { @Autowired private KafkaConsumer<String, String> kafkaConsumer; @RequestMapping("/receive") public List<String> receive() { 	////从Kafka服务器中的名称为"one-more-topic"的Topic中消费消息  ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofSeconds(1));  List<String> messages = new ArrayList<>(records.count());  for (ConsumerRecord<String, String> record : records.records("one-more-topic")) {   String message = record.value();   log.info("message: {}", message);   messages.add(message);  }  return messages; }}

文章持续更新，微信搜索「万猫学社」第一时间阅读，关注后回复「电子书」，免费获取12本Java必读技术书籍。

方式二：spring-kafka

使用kafka-clients需要我们自己创建生产者或者消费者的bean，如果我们的项目基于SpringBoot构建，那么使用spring-kafka就方便多了。

文章持续更新，微信搜索「万猫学社」第一时间阅读，关注后回复「电子书」，免费获取12本Java必读技术书籍。

引入依赖

在pom.

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

文章持续更新，微信搜索「万猫学社」第一时间阅读，关注后回复「电子书」，免费获取12本Java必读技术书籍。

生产者

在application.yml文件中增加配置：

spring: kafka: 	#Kafka服务器地址 bootstrap-servers: 127.0.0.1:9092 producer:  #设置数据value的序列化处理类  value-serializer: org.apache.kafka.common.serialization.StringSerializer

文章持续更新，微信搜索「万猫学社」第一时间阅读，关注后回复「电子书」，免费获取12本Java必读技术书籍。

在Controller中注入KafkaTemplate就可以直接使用了，代码如下：

@RestController@Slf4jpublic class Controller { @Autowired private KafkaTemplate<String, String> template; @RequestMapping("/springKafkaSend") public String send() {  String uuid = UUID.randomUUID().toString();  //将消息发送到Kafka服务器的名称为"one-more-topic"的Topic中  this.template.send("one-more-topic", uuid);  log.info("uuid: {}", uuid);  return uuid; }}

文章持续更新，微信搜索「万猫学社」第一时间阅读，关注后回复「电子书」，免费获取12本Java必读技术书籍。

消费者

在application.yml文件中增加配置：

spring: kafka: #Kafka服务器地址 bootstrap-servers: 127.0.0.1:9092 consumer:  #设置数据value的反序列化处理类  value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

文章持续更新，微信搜索「万猫学社」第一时间阅读，关注后回复「电子书」，免费获取12本Java必读技术书籍。

创建一个可以被Spring框架扫描到的类，并且在方法上加上@KafkaListener注解，就可以消费消息了，代码如下：

@Component@Slf4jpublic class Receiver { @KafkaListener(topics = "one-more-topic", groupId = "spring-kafka-group") public void listen(ConsumerRecord<?, ?> record) {  Optional<?> kafkaMessage = Optional.ofNullable(record.value());  if (kafkaMessage.isPresent()) {   String message = (String) kafkaMessage.get();   log.info("message: {}", message);  } }}

微信公众号：万猫学社

微信扫描二维码

关注后回复「电子书」

获取12本Java必读技术书籍

原文转载：http://www.shaoqun.com/a/505909.html

跨境电商：https://www.ikjzd.com/

myyearbook：https://www.ikjzd.com/w/726

邓白氏集团：https://www.ikjzd.com/w/582

Java实现Kafka生产者和消费者的示例Kafka简介Kafka是由Apache软件基金会开发的一个开源流处理平台，由Scala和Java编写。Kafka的目标是为处理实时数据提供一个统一、高吞吐、低延迟的平台。文章持续更新，微信搜索「万猫学社」第一时间阅读，关注后回复「电子书」，免费获取12本Java必读技术书籍。方式一：kafka-clients引入依赖在pom.<dependency
拍拍网：拍拍网
retriever：retriever
深圳世界之窗啤酒节有什么活动？2020世界之窗啤酒节游玩攻：深圳世界之窗啤酒节有什么活动？2020世界之窗啤酒节游玩攻
亚马逊关键词工具Ubersuggest：亚马逊关键词工具Ubersuggest
2020深圳锦绣中华民俗村端午节门票多少钱？锦绣中华端午节：2020深圳锦绣中华民俗村端午节门票多少钱？锦绣中华端午节