在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 

相关文章： 如何快速搭建高效可靠的建站解决方案？  中山网站制作网页,中山新生登记系统登记流程？  制作网站外包平台,自动化接单网站有哪些？  如何快速上传建站程序避免常见错误？  内部网站制作流程,如何建立公司内部网站？  常州自助建站工具推荐：低成本搭建与模板选择技巧  制作旅游网站html,怎样注册旅游网站？  成都品牌网站制作公司,成都营业执照年报网上怎么办理？  C++用Dijkstra(迪杰斯特拉)算法求最短路径  如何在景安云服务器上绑定域名并配置虚拟主机？  建站主机功能解析：服务器选择与快速搭建指南  如何零基础在云服务器搭建WordPress站点？  香港服务器租用每月最低只需15元？  建站之星免费版是否永久可用？  如何快速上传自定义模板至建站之星？  外贸公司网站制作,外贸网站建设一般有哪些步骤？  建站之星安装失败：服务器环境不兼容？  如何通过商城免费建站系统源码自定义网站主题？  如何用AWS免费套餐快速搭建高效网站？  如何注册花生壳免费域名并搭建个人网站？  电商网站制作价格怎么算,网上拍卖流程以及规则？  建站之星备案是否影响网站上线时间？  怎么制作网站设计模板图片,有电商商品详情页面的免费模板素材网站推荐吗？  c# 服务器GC和工作站GC的区别和设置  如何通过wdcp面板快速创建网站？  个人网站制作流程图片大全,个人网站如何注销？  青岛网站设计制作公司,查询青岛招聘信息的网站有哪些？  如何快速生成专业多端适配建站电话？  建站之星图片链接生成指南：自助建站与智能设计教程  建站主机类型有哪些？如何正确选型  建站之星如何保障用户数据免受黑客入侵？  如何做静态网页,sublimetext3.0制作静态网页？  建站之星Pro快速搭建教程：模板选择与功能配置指南  活动邀请函制作网站有哪些,活动邀请函文案？  建站一年半SEO优化实战指南：核心词挖掘与长尾流量提升策略  c# Task.Yield 的作用是什么 它和Task.Delay(1)有区别吗  如何打造高效商业网站？建站目的决定转化率  如何用花生壳三步快速搭建专属网站？  C#如何在一个XML文件中查找并替换文本内容  如何通过VPS建站实现广告与增值服务盈利？  教育培训网站制作流程,请问edu教育网站的域名怎么申请？  建站之星后台密码遗忘？如何快速找回？  教学网站制作软件,学习*后期制作的网站有哪些？  零服务器AI建站解决方案：快速部署与云端平台低成本实践  宝塔建站后网页无法访问如何解决？  建站之星如何助力企业快速打造五合一网站？  小视频制作网站有哪些,有什么看国内小视频的网站，求推荐？  网站设计制作企业有哪些,抖音官网主页怎么设置？  网站制作软件有哪些,制图软件有哪些？  建站主机如何安装配置？新手必看操作指南