当前位置: 首页 > news >正文

Spring-Kafka笔记整理

  1. 引入依赖
    <dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId>
    </dependency>
    
  2. 配置application.properties
    spring.kafka.bootstrap-servers=192.168.99.51:9092
    
  3. 编写kafka的配置类
    @Configuration
    public class KafkaConfig {@Value("${spring.kafka.bootstrap-servers}")private String bootstrapServers;@Beanpublic ProducerFactory<String, String> producerFactory() {Map<String, Object> configs = new HashMap<>();configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);return new DefaultKafkaProducerFactory<>(configs);}@Beanpublic KafkaTemplate<String, String> kafkaTemplate() {return new KafkaTemplate<>(producerFactory());}@Beanpublic ConsumerFactory<String, String> consumerFactory() {Map<String, Object> props = new HashMap<>();props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 50);props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);return new DefaultKafkaConsumerFactory<>(props);}@Beanpublic ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {ConcurrentKafkaListenerContainerFactory<String, String> factory =new ConcurrentKafkaListenerContainerFactory<>();// 并发数就是一个消费者实例起几个线程factory.setConcurrency(3);factory.setConsumerFactory(consumerFactory());return factory;}
    }
    
  4. Kafka消息监听
    @Component
    public class KafkaConsumer {@Autowiredprivate ObjectMapper mapper;@KafkaListener(topics = {"hello-kafka-topic"},groupId = "hello-kafka-group",containerFactory = "kafkaListenerContainerFactory")public void listener01(ConsumerRecord<String, String> record) throws Exception {String key = record.key();String value = record.value();HelloMessage kafkaMessage = mapper.readValue(value, HelloMessage.class);log.info("in listener consume kafka message: [{}], [{}]", key, mapper.writeValueAsString(kafkaMessage));}
    }
    
  5. Kafka消息发送
    @Component
    public class KafkaProducer {@Autowiredprivate KafkaTemplate<String, String> kafkaTemplate;public void sendMessage(String key, String value, String topic) {if (StringUtils.isBlank(value) || StringUtils.isBlank(topic)) {throw new IllegalArgumentException("value or topic is null or empty");}ListenableFuture<SendResult<String, String>> future = StringUtils.isBlank(key) ?kafkaTemplate.send(topic, value) : kafkaTemplate.send(topic, key, value);// 异步回调的方式获取通知future.addCallback(success -> {assert null != success && null != success.getRecordMetadata();// 发送到 kafka 的 topicString _topic = success.getRecordMetadata().topic();// 消息发送到的分区int partition = success.getRecordMetadata().partition();// 消息在分区内的 offsetlong offset = success.getRecordMetadata().offset();log.info("send kafka message success: [{}], [{}], [{}]", _topic, partition, offset);}, failure -> {log.error("send kafka message failure: [{}], [{}], [{}]", key, value, topic);});}
    }
    

相关文章:

Spring-Kafka笔记整理

引入依赖<dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId> </dependency>配置application.propertiesspring.kafka.bootstrap-servers192.168.99.51:9092编写kafka的配置类Configuration …...

已解决org.apache.hadoop.hdfs.protocol.QuotaExceededException异常的正确解决方法,亲测有效!!!

已解决org.apache.hadoop.hdfs.protocol.QuotaExceededException异常的正确解决方法&#xff0c;亲测有效&#xff01;&#xff01;&#xff01; 目录 问题分析 报错原因 解决思路 解决方法 总结 博主v&#xff1a;XiaoMing_Java 问题分析 在使用Hadoop分布式文件系统&a…...

GitHub打不开的解决方案(超简单)

在国内&#xff0c;github官网经常面临打不开或访问极慢的问题&#xff0c;不挂VPN&#xff08;梯子&#xff0c;飞机&#xff0c;魔法&#xff09;使用体验极差&#xff0c;那有什么好办法解决github官网访问不了的问题&#xff1f;今天小布教你几招轻松访问github官网。 git…...

Unity开发一个FPS游戏之二

在之前的文章中,我介绍了如何开发一个FPS游戏,添加一个第一人称的主角,并设置武器。现在我将继续完善这个游戏,打算添加敌人,实现其智能寻找玩家并进行对抗。完成的效果如下: fps_enemy_demo 下载资源 首先是设计敌人,我们可以在网上找到一些好的免费素材,例如在Unity…...

STM32F103 CubeMX 使用USB生成鼠标设备

STM32F103 CubeMX 使用USB生成鼠标设备 1 配置cubeMX1.1配置外部晶振&#xff0c;配置debug口1.2 配置USB1.3 配置芯片的时钟1.4 生成工程 2. 编写代码2.1 添加申明2.2 main函数代码 1 配置cubeMX 1.1配置外部晶振&#xff0c;配置debug口 1.2 配置USB 1.3 配置芯片的时钟 需…...

HJXH-E1/U静态信号继电器 面板安装 辅助电源220VDC 启动电压220VDC JOSEF约瑟

HJXH系列静态信号继电器 HJXH-61/U静态信号继电器&#xff1b; HJXH-61/I静态信号继电器&#xff1b; HJXH-62/U静态信号继电器&#xff1b; HJXH-62/I静态信号继电器&#xff1b; HJXH-E1/U静态信号继电器&#xff1b; HJXH-E1/I静态信号继电器&#xff1b; HJXH-E2/U静态信号…...

SpringBoot3下Kafka分组均衡消费实现

首先添加maven依赖&#xff1a; <dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId><version>2.8.11</version><exclusions><!--此处一定要排除kafka-clients&#xff0c;然…...

鸿蒙Harmony应用开发—ArkTS声明式开发(容器组件:GridItem)

网格容器中单项内容容器。 说明&#xff1a; 该组件从API Version 7开始支持。后续版本如有新增内容&#xff0c;则采用上角标单独标记该内容的起始版本。仅支持作为Grid组件的子组件使用。 子组件 可以包含单个子组件。 接口 GridItem GridItem(value?: GridItemOptions)…...

Qt 使用RAW INPUT获取HID触摸屏,笔设备,鼠标的原始数据,最低受支持的客户端:Windows XP [仅限桌面应用]

在开发绘图应用程序时&#xff0c;经常会需要读取笔设备的数据&#xff0c;通过对笔数据的解析&#xff0c;来判断笔的坐标&#xff0c;粗细。如果仅仅只是读取鼠标的坐标&#xff0c;就需要人为在应用程序端去修改笔的粗细&#xff0c;并且使用体验不好&#xff0c;如果可以实…...

easyexcel导出excel文件到s3服务器

导出excel文件是开发中常见的需求 常见的做法一般是直接通过请求接口响应对象HttpServletResponse把文件输出 我们可以使用原生的poi工具类操作.也可以使用easypoi.easyexcel等基于poi二次封装的工具处理 下面是代码 /*** 导出列表** param request* param response*/Overri…...

xss.haozi.me靶场“0x0B-0x12”通关教程

君衍. 一、0x0B 实体编码绕过二、0x0C script绕过三、0x0D 注释绕过四、0X0E ſ符号绕过五、0x0F 编码解码六、0x10 直接执行七、0x11 闭合绕过八、0x12 闭合绕过 XSS-Labs靶场“1-5”关通关教程 XSS-Labs靶场“6-10”关通关教程 Appcms存储型XSS漏洞复现 XSS-Labs靶场“11-13、…...

linux--redhat系统Yum源配置

1&#xff09;说明 redhat yum命令使用报错解决-重新配置yum源 解决--更改yum源 2&#xff09;更改yum源 &#xff08;1&#xff09;进入源目录 cd /etc/yum.repos.d/ &#xff08;2&#xff09;备份 redhat 默认源 mv redhat.repo redhat.repo-bak &#xff08;3&#xff09;…...

el-Switch 开关二次确认

前言 最近在做毕设&#xff0c;有个需求是点击按钮控制用户的状态是否禁用&#xff0c;就看到element有个switch组件可以改造一下&#xff0c;就上网看了一下&#xff0c;结果为了这个效果忙活了很久。。。所以说记录一下&#xff0c;让大家少踩坑。 前置条件 先看完我的需求再…...

(二)丶RabbitMQ的六大核心

一丶什么是MQ Message Queue(消息队列&#xff09;简称MQ&#xff0c;是一种应用程序对应用程序的消息通信机制。在MQ中&#xff0c;消息以队列形式存储&#xff0c;以便于异步传输&#xff0c;在MQ中&#xff0c;发布者&#xff08;生产者&#xff09;将消息放入队列&#xff…...

微信小程序实现上下手势滑动切换

效果图 思路 实现一个微信小程序的复合滚动页面&#xff0c;主要通过Swiper组件实现垂直方向的轮播功能&#xff0c;每个轮播项内部使用Scroll-View组件来展示可垂直滚动的长内容&#xff0c;如图片和文本。 代码 <!-- wxml --> <view class"swiper-container…...

详解命令docker run -d --name container_name -e TZ=Asia/Shanghai your_image

docker run 是Docker的主要命令&#xff0c;用于从镜像启动一个新的容器。下面详细解释并举例说明 -d, --name, -e TZ 参数的用法&#xff1a; -d 或 --detach&#xff1a; 这个标志告诉Docker以守护进程&#xff08;后台&#xff09;模式运行容器。这意味着当你执行 docker ru…...

javaEE7

1. <% page pageEncoding"UTF-8"%><% page import"java.io.*"%> <% page import"java.util.*"%> <% page import"java.math.*"%> <html> <head><title>网站计数器</title></head&…...

int与integer的区别

int和integer都是用来表示整数的数据类型&#xff0c;但有一些细微的区别。 int是Java中的基本数据类型&#xff0c;它可以存储整数值。int类型在内存中占4个字节&#xff0c;范围为-2,147,483,648到2,147,483,647。int类型使用最频繁&#xff0c;因为它的存储空间较小&#x…...

Golang实现Redis分布式锁(Lua脚本+可重入+自动续期)

Golang实现Redis分布式锁&#xff08;Lua脚本可重入自动续期&#xff09; 1 概念 应用场景 Golang自带的Lock锁单机版OK&#xff08;存储在程序的内存中&#xff09;&#xff0c;分布式不行 分布式锁&#xff1a; 简单版&#xff1a;redis setnx》加锁设置过期时间需要保证原…...

音乐播放器-C#实现

音乐播放器-C#实现 目录 一、 代码介绍 二、 音乐播放器-C#实现 三、 音乐播放器-C#实现 四、 音乐播放器-C#实现 五、 音乐播放器-C#实现 代码介绍 代码中使用了.NET框架中的System.Media命名空间来处理音频文件的播放和控制。这段代码创建了一个简单的音乐播放器界…...

Spring数据访问模块设计

前面我们已经完成了IoC和web模块的设计&#xff0c;聪明的码友立马就知道了&#xff0c;该到数据访问模块了&#xff0c;要不就这俩玩个6啊&#xff0c;查库势在必行&#xff0c;至此&#xff0c;它来了。 一、核心设计理念 1、痛点在哪 应用离不开数据&#xff08;数据库、No…...

laravel8+vue3.0+element-plus搭建方法

创建 laravel8 项目 composer create-project --prefer-dist laravel/laravel laravel8 8.* 安装 laravel/ui composer require laravel/ui 修改 package.json 文件 "devDependencies": {"vue/compiler-sfc": "^3.0.7","axios": …...

微软PowerBI考试 PL300-在 Power BI 中清理、转换和加载数据

微软PowerBI考试 PL300-在 Power BI 中清理、转换和加载数据 Power Query 具有大量专门帮助您清理和准备数据以供分析的功能。 您将了解如何简化复杂模型、更改数据类型、重命名对象和透视数据。 您还将了解如何分析列&#xff0c;以便知晓哪些列包含有价值的数据&#xff0c;…...

Spring是如何解决Bean的循环依赖:三级缓存机制

1、什么是 Bean 的循环依赖 在 Spring框架中,Bean 的循环依赖是指多个 Bean 之间‌互相持有对方引用‌,形成闭环依赖关系的现象。 多个 Bean 的依赖关系构成环形链路,例如: 双向依赖:Bean A 依赖 Bean B,同时 Bean B 也依赖 Bean A(A↔B)。链条循环: Bean A → Bean…...

JVM 内存结构 详解

内存结构 运行时数据区&#xff1a; Java虚拟机在运行Java程序过程中管理的内存区域。 程序计数器&#xff1a; ​ 线程私有&#xff0c;程序控制流的指示器&#xff0c;分支、循环、跳转、异常处理、线程恢复等基础功能都依赖这个计数器完成。 ​ 每个线程都有一个程序计数…...

Chromium 136 编译指南 Windows篇:depot_tools 配置与源码获取(二)

引言 工欲善其事&#xff0c;必先利其器。在完成了 Visual Studio 2022 和 Windows SDK 的安装后&#xff0c;我们即将接触到 Chromium 开发生态中最核心的工具——depot_tools。这个由 Google 精心打造的工具集&#xff0c;就像是连接开发者与 Chromium 庞大代码库的智能桥梁…...

云原生周刊:k0s 成为 CNCF 沙箱项目

开源项目推荐 HAMi HAMi&#xff08;原名 k8s‑vGPU‑scheduler&#xff09;是一款 CNCF Sandbox 级别的开源 K8s 中间件&#xff0c;通过虚拟化 GPU/NPU 等异构设备并支持内存、计算核心时间片隔离及共享调度&#xff0c;为容器提供统一接口&#xff0c;实现细粒度资源配额…...

实战设计模式之模板方法模式

概述 模板方法模式定义了一个操作中的算法骨架&#xff0c;并将某些步骤延迟到子类中实现。模板方法使得子类可以在不改变算法结构的前提下&#xff0c;重新定义算法中的某些步骤。简单来说&#xff0c;就是在一个方法中定义了要执行的步骤顺序或算法框架&#xff0c;但允许子类…...

Windows 下端口占用排查与释放全攻略

Windows 下端口占用排查与释放全攻略​ 在开发和运维过程中&#xff0c;经常会遇到端口被占用的问题&#xff08;如 8080、3306 等常用端口&#xff09;。本文将详细介绍如何通过命令行和图形化界面快速定位并释放被占用的端口&#xff0c;帮助你高效解决此类问题。​ 一、准…...

Python的__call__ 方法

在 Python 中&#xff0c;__call__ 是一个特殊的魔术方法&#xff08;magic method&#xff09;&#xff0c;它允许一个类的实例像函数一样被调用。当你在一个对象后面加上 () 并执行时&#xff08;例如 obj()&#xff09;&#xff0c;Python 会自动调用该对象的 __call__ 方法…...