在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

生产者日志：

消费者日志：

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

# springboot  # kafka  # 读写  # 滴滴二面之Kafka如何读写副本消息的  # 实现了  # 很好  # 只需  # 要在  # 大家多多  # 进行测试  # core  # context  # Configuration  # Bean  # util  # HashMap  # Map  # DefaultKafkaProducerFactory  # KafkaTemplate  # ProducerFactory  # annotation  # clients  # ProducerConfig  # apache 

相关文章： Python如何创建带属性的XML节点  高性能网站服务器配置指南：安全稳定与高效建站核心方案  保定网站制作方案定制,保定招聘的渠道有哪些？找工作的人一般都去哪里看招聘信息？  如何在建站主机中优化服务器配置？  网站视频怎么制作,哪个网站可以免费收看好莱坞经典大片？  已有域名和空间，如何快速搭建网站？  独立制作一个网站多少钱,建立网站需要花多少钱？  如何用VPS主机快速搭建个人网站？  建站10G流量真的够用吗？如何应对访问高峰？  实例解析angularjs的filter过滤器  建站主机核心功能解析：服务器选择与网站搭建流程指南  PHP正则匹配日期和时间(时间戳转换)的实例代码  香港网站服务器数量如何影响SEO优化效果？  建站中国官网：模板定制+SEO优化+建站流程一站式指南  公司网站建设制作费用,想建设一个属于自己的企业网站，该如何去做？  建站主机与服务器功能差异如何区分？  香港服务器网站卡顿？如何解决网络延迟与负载问题？  建站168自助建站系统：快速模板定制与SEO优化指南  如何通过西部建站助手安装IIS服务器？  家庭服务器如何搭建个人网站？  建站主机CVM配置优化、SEO策略与性能提升指南  山东云建站价格为何差异显著？  手机钓鱼网站怎么制作视频,怎样拦截钓鱼网站。怎么办？  香港服务器租用费用高吗？如何避免常见误区？  C++ static_cast和dynamic_cast区别_C++静态转换与动态类型安全转换  建站之星价格显示格式升级，你的预算足够吗？  SQL查询语句优化的实用方法总结  如何规划企业建站流程的关键步骤？  如何访问已购建站主机并解决登录问题？  焦点电影公司作品,电影焦点结局是什么？  上海制作企业网站有哪些,上海有哪些网站可以让企业免费发布招聘信息？  网站制作话术技巧,网站推广做的好怎么话术？  广东专业制作网站有哪些,广东省能源集团有限公司官网？  小捣蛋自助建站系统：数据分析与安全设置双核驱动网站优化  如何在Golang中引入测试模块_Golang测试包导入与使用实践  建站中国必看指南：CMS建站系统+手机网站搭建核心技巧解析  如何制作算命网站,怎么注册算命网站？  北京网站制作费用多少,建立一个公司网站的费用.有哪些部分,分别要多少钱？  如何选择CMS系统实现快速建站与SEO优化？  如何在IIS管理器中快速创建并配置网站？  建站之星备案流程有哪些注意事项？  如何快速生成凡客建站的专业级图册？  微信h5制作网站有哪些,免费微信H5页面制作工具？  大连网站制作费用,大连新青年网站，五年四班里的视频怎样下载啊？  制作农业网站的软件,比较好的农业网站推荐一下？  如何用IIS7快速搭建并优化网站站点？  建站VPS选购需注意哪些关键参数？  如何选择高效便捷的WAP商城建站系统？  定制建站哪家更专业可靠？推荐榜单揭晓  如何选择长沙网站建站模板？H5响应式与品牌定制哪个更优？