AI 助理
备案控制台
开发者社区 微服务 文章 正文

springboot配置kafka生产者和消费者详解

2018-10-19 9709
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 在原有pom.xml依赖下新添加一下kafka依赖ar包 org.springframework.kafka spring-kafka 1.

在原有pom.xml依赖下新添加一下kafka依赖ar包

<!--kafka-->
		<dependency>
			<groupId>org.springframework.kafka</groupId>
			<artifactId>spring-kafka</artifactId>
			<version>1.1.1.RELEASE</version>
		</dependency>
		<dependency>
			<groupId>org.apache.kafka</groupId>
			<artifactId>kafka_2.10</artifactId>
			<version>0.10.0.1</version>
		</dependency>

application.properties：

#原始数据kafka读取
kafka.consumer.servers=IP:9092,IP:9092(kafka消费集群ip+port端口)
kafka.consumer.enable.auto.commit=true(是否自动提交)
kafka.consumer.session.timeout=20000(连接超时时间)
kafka.consumer.auto.commit.interval=100
kafka.consumer.auto.offset.reset=latest(实时生产，实时消费，不会从头开始消费)
kafka.consumer.topic=result(消费的topic)
kafka.consumer.group.id=test(消费组)
kafka.consumer.concurrency=10(设置消费线程数)

#协议转换后存储kafka
kafka.producer.servers=IP:9092,IP:9092(kafka生产集群ip+port端口)
kafka.producer.topic=result(生产的topic)
kafka.producer.retries=0
kafka.producer.batch.size=4096
kafka.producer.linger=1
kafka.producer.buffer.memory=40960

springboot生产者配置：

package com.mapbar.track_storage.config;

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.annotation.EnableKafka;
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;

/**
 * kafka生产配置
 * @author Lvjiapeng
 *
 */
@Configuration
@EnableKafka
public class KafkaProducerConfig {
	@Value("${kafka.producer.servers}")
    private String servers;
    @Value("${kafka.producer.retries}")
    private int retries;
    @Value("${kafka.producer.batch.size}")
    private int batchSize;
    @Value("${kafka.producer.linger}")
    private int linger;
    @Value("${kafka.producer.buffer.memory}")
    private int bufferMemory;
    
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
        props.put(ProducerConfig.RETRIES_CONFIG, retries);
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
        props.put(ProducerConfig.LINGER_MS_CONFIG, linger);
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        return props;
    }

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

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

springboot消费者配置：

package com.mapbar.track_storage.config;

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.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;

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

/**
 * kafka消费者配置
 * @author Lvjiapeng
 *
 */
@Configuration
@EnableKafka
public class KafkaConsumerConfig {

    @Value("${kafka.consumer.servers}")
    private String servers;
    @Value("${kafka.consumer.enable.auto.commit}")
    private boolean enableAutoCommit;
    @Value("${kafka.consumer.session.timeout}")
    private String sessionTimeout;
    @Value("${kafka.consumer.auto.commit.interval}")
    private String autoCommitInterval;
    @Value("${kafka.consumer.group.id}")
    private String groupId;
    @Value("${kafka.consumer.auto.offset.reset}")
    private String autoOffsetReset;
    @Value("${kafka.consumer.concurrency}")
    private int concurrency;
    
    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(concurrency);
        factory.getContainerProperties().setPollTimeout(1500);
        return factory;
    }

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


    public Map<String, Object> consumerConfigs() {
        Map<String, Object> propsMap = new HashMap<>();
        propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
        propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
        propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitInterval);
        propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeout);
        propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
        return propsMap;
    }
    /**
     * kafka监听
     * @return
     */
    @Bean
    public RawDataListener listener() {
        return new RawDataListener();
    }

}

生产者测试：

package com.mapbar.track_storage.controller;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;

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

@RequestMapping(value = "/kafka")
@Controller
public class ProducerController {
    @Autowired
    private KafkaTemplate kafkaTemplate;

    @RequestMapping(value = "/producer",method = RequestMethod.GET)
    public void consume(HttpServletRequest request, HttpServletResponse response) throws IOException{
        String value = "{\"code\":200,\"dataVersion\":\"17q1\",\"message\":\"\",\"id\":\"364f79f28eea48eefeca8c85477a10d3\",\"source\":\"didi\",\"tripList\":[{\"subTripList\":[{\"startTimeStamp\":1519879598,\"schemeList\":[{\"distance\":0.0,\"ids\":\"94666702,\",\"schemeId\":0,\"linkList\":[{\"score\":72,\"distance\":1,\"gpsList\":[{\"origLonLat\":\"116.321343,40.43242\",\"grabLonLat\":\"112.32312,40.32132\",\"timestamp\":1515149926000}]}]}],\"endTimeStamp\":1519879598,\"subTripId\":0},{\"startTimeStamp\":1519879727,\"schemeList\":[{\"distance\":1395.0,\"ids\":\"94666729,7298838,7291709,7291706,88613298,88613297,7297542,7297541,94698785,94698786,94698778,94698780,94698779,94698782,\",\"schemeId\":0,\"linkList\":[{\"score\":72,\"distance\":1,\"gpsList\":[{\"origLonLat\":\"116.321343,40.43242\",\"grabLonLat\":\"112.32312,40.32132\",\"timestamp\":1515149926000}]}]}],\"endTimeStamp\":1519879812,\"subTripId\":1},{\"startTimeStamp\":1519879836,\"schemeList\":[{\"distance\":0.0,\"ids\":\"54123007,\",\"schemeId\":0,\"linkList\":[{\"score\":72,\"distance\":1,\"gpsList\":[{\"origLonLat\":\"116.321343,40.43242\",\"grabLonLat\":\"112.32312,40.32132\",\"timestamp\":1515149926000}]}]}],\"endTimeStamp\":1519879904,\"subTripId\":2},{\"startTimeStamp\":1519879959,\"schemeList\":[{\"distance\":0.0,\"ids\":\"54190443,\",\"schemeId\":0,\"linkList\":[{\"score\":72,\"distance\":1,\"gpsList\":[{\"origLonLat\":\"116.321343,40.43242\",\"grabLonLat\":\"112.32312,40.32132\",\"timestamp\":1515149926000}]}]}],\"endTimeStamp\":1519879959,\"subTripId\":3},{\"startTimeStamp\":1519880088,\"schemeList\":[{\"distance\":2885.0,\"ids\":\"94698824,94698822,94698789,94698786,54123011,54123012,54123002,94698763,94698727,94698722,94698765,54123006,54123004,\",\"schemeId\":0,\"linkList\":[{\"score\":72,\"distance\":1,\"gpsList\":[{\"origLonLat\":\"116.321343,40.43242\",\"grabLonLat\":\"112.32312,40.32132\",\"timestamp\":1515149926000}]}]}],\"endTimeStamp\":1519880300,\"subTripId\":4},{\"startTimeStamp\":1519880393,\"schemeList\":[{\"distance\":2398.0,\"ids\":\"7309441,7303680,54123061,54123038,7309478,7309477,94698204,94698203,94698273,94698274,94698288,94698296,94698295,94698289,94698310,\",\"schemeId\":0,\"linkList\":[{\"score\":72,\"distance\":1,\"gpsList\":[{\"origLonLat\":\"116.321343,40.43242\",\"grabLonLat\":\"112.32312,40.32132\",\"timestamp\":1515149926000}]}]}],\"endTimeStamp\":1519880636,\"subTripId\":5},{\"startTimeStamp\":1519881064,\"schemeList\":[{\"distance\":35.0,\"ids\":\"7309474,\",\"schemeId\":0,\"linkList\":[{\"score\":72,\"distance\":1,\"gpsList\":[{\"origLonLat\":\"116.321343,40.43242\",\"grabLonLat\":\"112.32312,40.32132\",\"timestamp\":1515149926000}]}]}],\"endTimeStamp\":1519881204,\"subTripId\":6},{\"startTimeStamp\":1519881204,\"schemeList\":[{\"distance\":28.0,\"ids\":\"7309476,\",\"schemeId\":0,\"linkList\":[{\"score\":72,\"distance\":1,\"gpsList\":[{\"origLonLat\":\"116.321343,40.43242\",\"grabLonLat\":\"112.32312,40.32132\",\"timestamp\":1515149926000}]}]}],\"endTimeStamp\":1519881266,\"subTripId\":7},{\"startTimeStamp\":1519881291,\"schemeList\":[{\"distance\":463.0,\"ids\":\"7303683,\",\"schemeId\":0,\"linkList\":[{\"score\":72,\"distance\":1,\"gpsList\":[{\"origLonLat\":\"116.321343,40.43242\",\"grabLonLat\":\"112.32312,40.32132\",\"timestamp\":1515149926000}]}]}],\"endTimeStamp\":1519881329,\"subTripId\":8}],\"startTimeStamp\":1519879350,\"unUseTime\":1201,\"totalTime\":2049,\"endTimeStamp\":1519881399,\"tripId\":0}]}";
        for (int i = 1; i<=500; i++){
            kafkaTemplate.send("result",value);
        }
    }
}

消费者测试：

import net.sf.json.JSONObject;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.List;

/**
 * kafka监听
 * @author shangzz
 *
 */
@Component
public class RawDataListener {
	Logger logger=Logger.getLogger(RawDataListener.class);
	@Autowired
	private MatchRoadService matchRoadService;

	/**
	 * 实时获取kafka数据(生产一条，监听生产topic自动消费一条)
	 * @param record
	 * @throws IOException
	 */
	@KafkaListener(topics = {"${kafka.consumer.topic}"})
    public void listen(ConsumerRecord<?, ?> record) throws IOException {
		String value = (String) record.value();
		System.out.println(value);
	}

}

总结：

         ①  生产者环境类配置好以后，@Autowired自动注入KafkaTemplate类，使用send方法生产消息

         ②  消费者环境类配置好以后，方法头前使用@KafkaListener(topics = {"${kafka.consumer.topic}"})注解监听topic并传入ConsumerRecord<?, ?> record对象即可自动消费topic

         ③  相关kafka配置只需在application.properties照葫芦画瓢添加，修改或者删除配置并在环境配置类中做出相应修改即可

文章标签：
云消息队列 Kafka 版
消息中间件
Kafka
Apache
Java
关键词：
云消息队列 Kafka 版配置
云消息队列 Kafka 版消费者
云消息队列 Kafka 版生产者
配置云消息队列 Kafka 版
云消息队列 Kafka 版springboot
jiapeng_lv
目录
相关文章
kaixin321-44007
|
20天前
|
消息中间件 存储 Prometheus
Kafka集群如何配置高可用性
Kafka集群如何配置高可用性
kaixin321-44007
36 1
武子康
|
1月前
|
消息中间件 监控 Ubuntu
大数据-54 Kafka 安装配置 环境变量配置 启动服务 Ubuntu配置 ZooKeeper
大数据-54 Kafka 安装配置 环境变量配置 启动服务 Ubuntu配置 ZooKeeper
武子康
77 3
大数据-54 Kafka 安装配置 环境变量配置 启动服务 Ubuntu配置 ZooKeeper
郑小健
|
22天前
|
消息中间件 存储 负载均衡
Apache Kafka核心概念解析：生产者、消费者与Broker
【10月更文挑战第24天】在数字化转型的大潮中，数据的实时处理能力成为了企业竞争力的重要组成部分。Apache Kafka 作为一款高性能的消息队列系统，在这一领域占据了重要地位。通过使用 Kafka，企业可以构建出高效的数据管道，实现数据的快速传输和处理。今天，我将从个人的角度出发，深入解析 Kafka 的三大核心组件——生产者、消费者与 Broker，希望能够帮助大家建立起对 Kafka 内部机制的基本理解。
郑小健
52 2
武子康
|
1月前
|
消息中间件 分布式计算 Java
大数据-73 Kafka 高级特性 稳定性-事务 相关配置 事务操作Java 幂等性 仅一次发送
大数据-73 Kafka 高级特性 稳定性-事务 相关配置 事务操作Java 幂等性 仅一次发送
武子康
31 2
武子康
|
1月前
|
消息中间件 SQL 分布式计算
大数据-76 Kafka 高级特性 稳定性-消费重复 生产者、Broker、消费者 导致的重复消费问题
大数据-76 Kafka 高级特性 稳定性-消费重复 生产者、Broker、消费者 导致的重复消费问题
武子康
34 1
武子康
|
1月前
|
消息中间件 Java 大数据
大数据-56 Kafka SpringBoot与Kafka 基础简单配置和使用 Java代码 POM文件
大数据-56 Kafka SpringBoot与Kafka 基础简单配置和使用 Java代码 POM文件
武子康
65 2
游客j4mujezz7vm7y
|
2月前
|
消息中间件 Kafka
消费kafka不需要设置 压缩协议吗 假如生产者压缩协议是lz4
消费kafka不需要设置 压缩协议吗 假如生产者压缩协议是lz4
游客j4mujezz7vm7y
76 7
武子康
|
1月前
|
消息中间件 NoSQL Kafka
大数据-116 - Flink DataStream Sink 原理、概念、常见Sink类型 配置与使用 附带案例1：消费Kafka写到Redis
大数据-116 - Flink DataStream Sink 原理、概念、常见Sink类型 配置与使用 附带案例1：消费Kafka写到Redis
武子康
133 0
土木林森
|
3月前
|
消息中间件 开发框架 Java
掌握这一招，Spring Boot与Kafka完美融合，顺序消费不再是难题，让你轻松应对业务挑战！
【8月更文挑战第29天】Spring Boot与Kafka集成广泛用于处理分布式消息队列。本文探讨了在Spring Boot中实现Kafka顺序消费的方法，包括使用单个Partition或消息Key确保消息路由到同一Partition，并设置Consumer并发数为1以保证顺序消费。通过示例代码展示了如何配置Kafka Producer和Consumer，并自定义Partitioner。为确保数据正确性，还建议在业务逻辑中增加顺序校验机制。
土木林森
105 3
土木林森
|
3月前
|
消息中间件 Kafka Java
Spring 框架与 Kafka 联姻，竟引发软件世界的革命风暴！事件驱动架构震撼登场！
【8月更文挑战第31天】《Spring 框架与 Kafka 集成：实现事件驱动架构》介绍如何利用 Spring 框架的强大功能与 Kafka 分布式流平台结合，构建灵活且可扩展的事件驱动系统。通过添加 Spring Kafka 依赖并配置 Kafka 连接信息，可以轻松实现消息的生产和消费。文中详细展示了如何设置 `KafkaTemplate`、`ProducerFactory` 和 `ConsumerFactory`，并通过示例代码说明了生产者发送消息及消费者接收消息的具体实现。这一组合为构建高效可靠的分布式应用程序提供了有力支持。
土木林森
109 0

热门文章

最新文章

  • 1
    解决kafka集群由于默认的__consumer_offsets这个topic的默认的副本数为1而存在的单点故障问题
  • 2
    SQLServer CDC数据通过Kafka connect实时同步至分析型数据库 AnalyticDB For PostgreSQL及OSS
  • 3
    阿里云Kafka幂等生产者与事务生产者
  • 4
    Apache Kafka开发入门指南
  • 5
    Kafka Producer 异步发送消息居然也会阻塞？
  • 6
    kafka集群搭建
  • 7
    ELK+Kafka 企业日志收集平台(一)
  • 8
    kafka web console安装
  • 9
    如何查看spark与hadoop、kafka、Scala、flume、hive等兼容版本【适用于任何版本】
  • 10
    【Kafka】（二十）Kafka Consumer 重置 Offset
  • 1
    微信小程序和springboot实现微信支付
    226
  • 2
    springboot用户创建的业务数据只能是同一组织能看的见
    35
  • 3
    深入浅出Redis（十三）：SpringBoot整合Redis客户端
    271
  • 4
    保姆级教程——将springboot项目部署到阿里云服务器包含环境配置（小白包会）
    1020
  • 5
    深入理解Spring Boot启动流程及其实战应用
    190
  • 6
    Java医院绩效考核系统源码B/S架构+springboot三级公立医院绩效考核系统源码 医院综合绩效核算系统源码
    86
  • 7
    springboot基层区域HIS系统源码
    50
  • 8
    springboot之RedisTemplate的访问单机，哨兵，集群模式
    392
  • 9
    在IntelliJ IDEA中通过Spring Boot集成达梦数据库：从入门到精通
    1418
  • 10
    SpringBoot 拦截器 统一结果返回 统一异常处理
    319

    • 相关课程

    更多
  • SpringBoot实战教程
  • 精通Spring Cloud Alibaba
  • SpringBoot快速掌握 - 核心技术
  • SpringBoot快速掌握 - 高级应用
  • 分布式消息系统 Kafka 快速入门
  • Springboot项目云开发快速迁移

    • 相关电子书

    更多
  • Java Spring Boot开发实战系列课程【第16讲】：Spring Boot 2.0 实战Apache Kafka百万级高并发消息中间件与原理解析
  • MaxCompute技术公开课第四季 之 如何将Kafka数据同步至MaxCompute
  • 消息队列kafka介绍

    • 相关实验场景

    更多
  • 快速体验云消息队列RocketMQ版
  • 消息队列RocketMQ版：基础消息收发功能体验
  • 快速学会使用云消息队列RabbitMQ版
  • 消息队列RocketMQ版：消费异常运维排查体验
  • 消息队列RocketMQ版：定时消息通知功能体验
  • 部署RabbitMQ
    • 下一篇
    阿里云OSS设置跨域访问