脚本之家 服务器常用软件
快捷导航
您的位置：首页 → 软件编程 → java → springboot kafka进行读写

在springboot中对kafka进行读写的示例代码

 更新时间：2017年09月06日 14:25:44   作者：冬天里的懒喵  
本篇文章主要介绍了在springboot中对kafka进行读写的示例代码，小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧

springboot对kafka的client很好的实现了集成，使用非常方便，本文也实现了一个在springboot中实现操作kafka的demo。

1.POM配置

只需要在dependencies中增加 spring-kafka的配置即可。完整效果如下：

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>1.5.4.RELEASE</version>
  </parent>

  <properties>
    <java.version>1.8</java.version>
     <spring-kafka.version>1.2.2.RELEASE</spring-kafka.version>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  </properties>

  <dependencies>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter</artifactId>
    </dependency>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-test</artifactId>
      <scope>test</scope>
    </dependency>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-aop</artifactId>
    </dependency>
   <!-- spring-kafka -->
      <dependency>
      <groupId>org.springframework.kafka</groupId>
      <artifactId>spring-kafka</artifactId>
      <version>${spring-kafka.version}</version>
      </dependency>
      <dependency>
      <groupId>org.springframework.kafka</groupId>
      <artifactId>spring-kafka-test</artifactId>
      <version>${spring-kafka.version}</version>
      <scope>test</scope>
      </dependency>
   </dependencies>

2.生产者

参数配置类，其参数卸载yml文件中，通过@Value注入

package com.dhb.kafka.producer;

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class SenderConfig {

  @Value("${kafka.bootstrap-servers}")
  private String bootstrapServers;

  @Bean
  public Map<String,Object> producerConfigs() {
    Map<String,Object> props = new HashMap<>();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,this.bootstrapServers);
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class);
    props.put(ProducerConfig.ACKS_CONFIG,"0");
    return props;
  }

  @Bean
  public ProducerFactory<String,String> producerFactory() {
    return new DefaultKafkaProducerFactory<>(producerConfigs());
  }

  @Bean
  public KafkaTemplate<String,String> kafkaTemplate() {
    return new KafkaTemplate<String, String>(producerFactory());
  }

  @Bean
  public Sender sender() {
    return new Sender();
  }
}

消息发送类

package com.dhb.kafka.producer;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;

@Slf4j
public class Sender {

  @Autowired
  private KafkaTemplate<String,String> kafkaTemplate;

  public void send(String topic,String payload) {
    log.info("sending payload='{}' to topic='{}'",payload,topic);
    this.kafkaTemplate.send(topic,payload);
  }
}

3.消费者

参数配置类

package com.dhb.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;

import java.util.HashMap;
import java.util.Map;

@Configuration
@EnableKafka
public class ReceiverConfig {

  @Value("${kafka.bootstrap-servers}")
  private String bootstrapServers;

  public Map<String,Object> consumerConfigs() {
    Map<String,Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,bootstrapServers);
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class);
    props.put(ConsumerConfig.GROUP_ID_CONFIG,"helloword");
    return props;
  }

  @Bean
  public ConsumerFactory<String,String> consumerFactory() {
    return new DefaultKafkaConsumerFactory<>(consumerConfigs());
  }

  @Bean
  public ConcurrentKafkaListenerContainerFactory<String,String> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String,String> factory =
        new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    return factory;
  }

  @Bean
  public Receiver receiver() {
    return new Receiver();
  }

}

消息接受类

package com.dhb.kafka.consumer;

import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;

import java.util.concurrent.CountDownLatch;

@Slf4j
public class Receiver {

  private CountDownLatch latch = new CountDownLatch(1);

  public CountDownLatch getLatch() {
    return latch;
  }

  @KafkaListener(topics = "${kafka.topic.helloworld}")
  public void receive(String payload) {
    log.info("received payload='{}'",payload);
    latch.countDown();
  }
}

3.web测试类

定义了一个基于http的web测试接口

package com.dhb.kafka.web;

import com.dhb.kafka.producer.Sender;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;

import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;

@RestController
@Slf4j
public class KafkaProducer {

  @Autowired
  Sender sender;

  @RequestMapping(value = "/sender.action", method = RequestMethod.POST)
  public void exec(HttpServletRequest request, HttpServletResponse response,String data) throws IOException{
    this.sender.send("testtopic",data);
    response.setCharacterEncoding("UTF-8");
    response.setContentType("text/json");
    response.getWriter().write("success");
    response.getWriter().flush();
    response.getWriter().close();
  }

}

4.启动类及配置

package com.dhb.kafka;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class KafkaApplication {


  public static void main(String[] args) {
    SpringApplication.run(KafkaApplication.class,args);

  }
}

application.yml

kafka:
 bootstrap-servers: 192.168.162.239:9092
 topic:
  helloworld: testtopic

程序结构：

包结构

5.读写测试

通过执行KafkaApplication的main方法启动程序。然后打开postman进行测试：


运行后返回success


生产者日志：


消费者日志：


以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持脚本之家。

您可能感兴趣的文章:

相关文章

  • IDEA 2021.2 激活教程及启动报错问题解决方法

    IDEA 2021.2 激活教程及启动报错问题解决方法

    这篇文章主要介绍了IDEA 2021.2 启动报错及激活教程,文章开头给大家介绍了idea2021最新激活方法，关于idea2021启动报错的问题小编也给大家介绍的非常详细，需要的朋友可以参考下
    2021-10-10
  • 完美解决springboot项目出现”java: 错误: 无效的源发行版:17“问题(图文详解)

    完美解决springboot项目出现”java: 错误: 无效的源发行版:17“问题(图文详解)

    这篇文章主要介绍了完美解决springboot项目出现”java: 错误: 无效的源发行版:17“问题,本文通过图文并茂的形式给大家介绍的非常详细，需要的朋友可以参考下
    2023-04-04
  • Mybatis 开发注解快速入门

    Mybatis 开发注解快速入门

    mybatis是一个支持普通SQL查询，存储过程和高级映射的优秀持久层框架。这篇文章主要介绍了Mybatis 开发注解快速入门的相关资料,需要的朋友可以参考下
    2016-11-11
  • Java 数据库时间返回前端显示错误(差8个小时)的解决方法

    Java 数据库时间返回前端显示错误(差8个小时)的解决方法

    本文主要介绍了Java 数据库时间返回前端显示错误(差8个小时)的解决方法,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧
    2023-08-08
  • jar中VO的探究

    jar中VO的探究

    这篇文章主要介绍了jar中VO的探究的相关资料,需要的朋友可以参考下
    2023-11-11
  • SpringBoot集成Lettuce客户端操作Redis的实现

    SpringBoot集成Lettuce客户端操作Redis的实现

    本文主要介绍了SpringBoot集成Lettuce客户端操作Redis的实现,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧
    2023-11-11
  • Mybatis映射文件规则实例详解

    Mybatis映射文件规则实例详解

    在映射文件当中,mapper元素是映射文件的根元素,其他的标签都是它的子元素,下面这篇文章主要给大家介绍了关于Mybatis映射文件规则的相关资料,需要的朋友可以参考下
    2022-04-04
  • Java实现将对象转换为二进制流再转为Base64字符串

    Java实现将对象转换为二进制流再转为Base64字符串

    这篇文章主要为大家详细介绍了如何使用Java实现将对象转换为二进制流再转为Base64字符串,本文提供了四种方法,大家可以根据需要进行选择
    2025-12-12
  • Java中常见的文件拷贝方式小结

    Java中常见的文件拷贝方式小结

    这篇文章主要为大家详细介绍了JAVA 四种拷贝文件的方式,分析一下他们对内存使用的方式和各自应用的场景,其实也是对之前学过的知识做一个回顾吧,快跟随小编一起学习起来吧
    2024-03-03
  • Java 中的 Consumer 与 Supplier 接口使用场景与示例

    Java 中的 Consumer 与 Supplier 接口使用场景与

    文章介绍了Java 8中函数式接口的核心概念,Consumer与Supplier接口使用场景与示例,重点对比了Consumer（消费数据）与Supplier（供给数据）的区别,感兴趣的朋友跟随小编一起看看吧
    2025-08-08

最新评论