准备工作

当然我们也可以不手动创建topic，在执行代码kafkaTemplate.send(“topic1”, normalMessage)发送消息时，kafka会帮我们自动完成topic的创建工作，但这种情况下创建的topic默认只有一个分区，分区也没有副本。所以，我们可以在项目中新建一个配置类专门用来初始化topic，如下

初始化配置信息

import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class KafkaInitialConfiguration {/**** 创建TopicName为topic.quick.initial的Topic并设置分区数为8以及副本数为1* 通过bean创建(bean的名字为initialTopic)* @return*/@Beanpublic NewTopic initialTopic() {return new NewTopic("topic.quick.initial",8, (short) 1 );}/*** 此种@Bean的方式，如果topic的名字相同，那么会覆盖以前的那个* //修改后|分区数量会变成11个 注意分区数量只能增加不能减少* @return*/@Beanpublic NewTopic initialTopic2() {return new NewTopic("topic.quick.initial",11, (short) 1 );}}

程序代码

生产者

@Slf4j
@RestController
@RequestMapping("/api/kafka")
@Api(tags = "kafka测试开发")
public class KafkaController {@Autowiredprivate KafkaTemplate<String, Object> kafkaTemplate;@Resource(name = "initialTopic")private NewTopic newTopic;@GetMapping("/callbackOne")@ApiOperation(value = "带回调的生产者")public void sendMessage2(@RequestParam("message") @ApiParam(value="消息",required = true) String callbackMessage) {log.info("========================================>>>");log.info(newTopic.name());// 带回调的生产者kafkaTemplate.send(newTopic.name(), callbackMessage).addCallback(success -> {// 消息发送到的topicString topic = success.getRecordMetadata().topic();// 消息发送到的分区int partition = success.getRecordMetadata().partition();// 消息在分区内的offsetlong offset = success.getRecordMetadata().offset();log.info("发送消息成功:" + topic + "-" + partition + "-" + offset + "-" + callbackMessage);}, failure -> {log.info("发送消息失败:" + failure.getMessage());});}}

消费者

@Component
@Slf4j
public class KafkaConsumer {// 消费监听@KafkaListener(topics = {"topic.quick.initial"})public void onMessage1(ConsumerRecord<?, ?> record){// 消费的哪个topic、partition的消息,打印出消息内容log.info("==============================================>");StringBuffer sb = new StringBuffer();// 主题sb.append(record.topic() + "-");// 分区sb.append(record.partition() + "-");// 需要消费的值sb.append(record.value() + "-");// 位移sb.append(record.offset());log.info("消费者进行消费："+ sb);}
}

程序测试

使用swagger来进行程序测试

本文小结

本文简单进行了SpringBoot整合kafka之kafka分区实战。

SpringBoot整合kafka之kafka分区实战相关推荐

搭建大型分布式服务（二十二）SpringBoot 如何优雅地整合多个kafka数据源？
系列文章目录文章目录系列文章目录前言一.本文要点二.开发环境三.创建项目四.修改项目五.测试一下六.小结前言在日常开发当中,经常会遇到需要消费的topic不在同一个kafka集群 ...
【七】springboot整合redis（超详细）
springboot篇章整体栏目: [一]springboot整合swagger(超详细 [二]springboot整合swagger(自定义)(超详细) [三]springboot整合token(超 ...
SpringBoot整合kafka实战之带回调的生产者
本文来说下SpringBoot整合kafka部分知识内容文章目录带回调的生产者方式一方式二本文小结带回调的生产者前面我们说了简单的生产和消费,本文说下带回调的生产者.kafkaTempl ...
springboot 整合 kafka demo 顺便看一下源码
大家好,我是烤鸭: 今天分享下 springboot 整合 kafka. 1. 环境参数: windows + kafka_2.11-2.3.0 + zookeeper-3.5.6 + spring ...
kafka自定义分区实战
本文来说下kafka自定义分区相关的知识与内容,同时说下springboot整合kafka如何来实现自定义分区文章目录 Kafka如何实现分区 Kafka集群是如何知道投递到哪个broker中默认 ...
SpringBoot整合kafka(安装)
项目路径:https://github.com/zhaopeng01/springboot-study/tree/master/study_14 序言 Kafka 是一种高吞吐的分布式发布订阅消息系统 ...
springboot 整合kafka 实现生产，消费数据
一 kafka集群的启动 1.1 机器说明 192.168.152.128 master 192.168.152.129 slaver01 192.168.152.130 slaver02 1.2 查 ...
Kafka精品教学（入门，安装，Springboot整合Kafka）
ps:本文是博主结合视频和博客学习之后,自己实验总结编写的,如果侵权请联系删除. 要学习kafka首先要了解什么是消息队列,因为Kafka 是一个分布式的基于发布 / 订阅模式的消息队列(Messag ...
Kafka原理以及SpringBoot整合Kafka
1.Kafka原理 1. brokers有多个broker组成,broker是指Kafka服务器(192.168.223.140就是其中的一个broker),上面三台Kafka服务器组成了Kafka集 ...

SpringBoot整合kafka之kafka分区实战

文章目录

准备工作

程序代码

程序测试

本文小结

SpringBoot整合kafka之kafka分区实战相关推荐

最新文章

热门文章