kafka-生产者事务-数据传递语义事务介绍事务消息发送（SpringBoot整合Kafka）

文章目录

  • 1、kafka数据传递语义
  • 2、kafka生产者事务
  • 3、事务消息发送
    • 3.1、application.yml配置
    • 3.2、创建生产者监听器
    • 3.3、创建生产者拦截器
    • 3.4、发送消息测试
    • 3.5、使用Java代码创建主题分区副本
    • 3.6、屏蔽 kafka debug 日志 logback.xml
    • 3.7、引入spring-kafka依赖
    • 3.8、控制台日志

1、kafka数据传递语义

kafka发送消息时是否需要重试

  1. 仅发送一次：生产者发送消息后不重试，只发送一次 可能丢失消息 效率最高
  2. 至少一次：生产者发送消息后重试，可能重试多次 效率差
  3. 精准一次发送：生产者发送消息后无论是否重复发送 发送了多少次，在 kafka broker 中只保存一次消息，通过幂等性 + 生产者事务来实现

kafak天然支持幂等性，每个消息头中带了一个唯一的标志 kafka broker 根据此标志判断消息是否已经发送过，生产者事务可以保证数据没有最终发送成功时，消费者不可以消费，如果生产者发送消息时出现异常会自动回滚（清除之前发送的事务中的消息）

kafka天然幂等性：但是指的是生产者事务 生产消息时的幂等性，发送消息时消息中带唯一标识、broker接收到消息时如果重复不再保存，事务没提交消费者不能消费改消息

2、kafka生产者事务

保证消息生产的幂等性
一组消息要么一起成功 被消费者消息 要么一起失败都不能被消费者消费

  1. 配置ack为-1 分区所有副本均落盘成功
  2. 配置生产者重试（发送失败可以继续发送：需要保证发送失败后再次发送消息到kafka实现 精准一次发送）
  3. 需要给事务分配事务id（区分一个事务中的多条消息）

3、事务消息发送

3.1、application.yml配置

server:
  port: 8110

# v1
spring:
  kafka:
    bootstrap-servers: 192.168.74.148:9095,192.168.74.148:9096,192.168.74.148:9097
    producer: # producer 生产者
      retries: 1 # 重试次数 0表示不重试
      acks: -1 # 应答级别:多少个分区副本备份完成时向生产者发送ack确认(可选0、1、-1/all)
      transaction-id-prefix: tx_  # 事务id前缀：配置后producer自动开启事务
      batch-size: 16384 # 批次大小 单位byte
      buffer-memory: 33554432 # 生产者缓冲区大小 单位byte
      key-serializer: org.apache.kafka.common.serialization.StringSerializer # key的序列化器
      value-serializer: org.apache.kafka.common.serialization.StringSerializer # value的序列化器

3.2、创建生产者监听器

package com.atguigu.kafka.listener;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.kafka.support.ProducerListener;
@Component
public class MyKafkaProducerListener implements ProducerListener<String,String> {

    //生产者 ack 配置为 0 只要发送即成功
    //ack为 1  leader落盘  broker ack之后 才成功
    //ack为 -1 分区所有副本全部落盘  broker ack之后 才成功
    @Override
    public void onSuccess(ProducerRecord<String, String> producerRecord, RecordMetadata recordMetadata) {
        //ProducerListener.super.onSuccess(producerRecord, recordMetadata);
        System.out.println("MyKafkaProducerListener消息发送成功："+"topic="+producerRecord.topic()
        +"，partition = "+producerRecord.partition()
        +"，key = "+producerRecord.key()
        +"，value = "+producerRecord.value()
        +"，offset = "+recordMetadata.offset());
    }

    //消息发送失败的回调：监听器可以接收到发送失败的消息 可以记录失败的消息
    @Override
    public void onError(ProducerRecord<String, String> producerRecord, RecordMetadata recordMetadata, Exception exception) {
        System.out.println("MyKafkaProducerListener消息发送失败："+"topic="+producerRecord.topic()
                +"，partition = "+producerRecord.partition()
                +"，key = "+producerRecord.key()
                +"，value = "+producerRecord.value()
                +"，offset = "+recordMetadata.offset());
        System.out.println("异常信息：" + exception.getMessage());
    }
}

3.3、创建生产者拦截器

package com.atguigu.kafka.interceptor;
import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.stereotype.Component;
import java.util.Map;
//拦截器必须手动注册给kafka生产者（KafkaTemplate）
@Component
public class MyKafkaInterceptor implements ProducerInterceptor<String,String> {
    //kafka生产者发送消息前执行：拦截发送的消息预处理
    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> producerRecord) {
        System.out.println("生产者即将发送消息：topic = "+ producerRecord.topic()
        +"，partition:"+producerRecord.partition()
        +"，key = "+producerRecord.key()
        +"，value = "+producerRecord.value());
        return null;
    }

    //kafka broker 给出应答后执行
    @Override
    public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {
        //exception为空表示消息发送成功
        if(e == null){
            System.out.println("消息发送成功：topic = "+ recordMetadata.topic()
                    +"，partition:"+recordMetadata.partition()
                    +"，offset="+recordMetadata.offset()
            +"，timestamp="+recordMetadata.timestamp());
        }
    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> map) {

    }
}

3.4、发送消息测试

package com.atguigu.kafka.producer;

import com.atguigu.kafka.interceptor.MyKafkaInterceptor;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.Resource;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.core.KafkaTemplate;
import java.io.IOException;

@SpringBootTest
class KafkaProducerApplicationTests {

    //装配kafka模板类： springboot启动时会自动根据配置文初始化kafka模板类对象注入到容器中
    @Resource
    KafkaTemplate kafkaTemplate;

    @Resource
    MyKafkaInterceptor myKafkaInterceptor;

    @PostConstruct
    public void init() {
        kafkaTemplate.setProducerInterceptor(myKafkaInterceptor);
    }
    @Test
    void contextLoads() throws IOException {
        kafkaTemplate.send("my_topic1", "spring-kafka-生产者监听器");
        //回调是等kafka，ack以后才执行，需要阻塞
        System.in.read();
    }

    //kafka事务支持spring-tx的事务注解
    //单元测试中的事务会自动回滚

    @Test
    void testTransaction() throws  IOException {

       //多个消息的发送在一个事务中执行
        kafkaTemplate.executeInTransaction((var1) -> {
            //通过一个事务中的operations对象来发送消息,执行事务操作
            var1.send("my_topic1",0,"", "spring-kafka-事务1");
            var1.send("my_topic1",0,"", "spring-kafka-事务2");
            int i = 1/0;
            var1.send("my_topic1",0,"", "spring-kafka-事务3");
            return "发送消息失败";
        });
        System.in.read();
    }
}

3.5、使用Java代码创建主题分区副本

package com.atguigu.kafka.config;
import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.config.TopicBuilder;
import org.springframework.stereotype.Component;
@Component
public class KafkaTopicConfig {
    @Bean
    public NewTopic myTopic1() {
        //相同名称的主题 只会创建一次，后面创建的主题名称相同配置不同可以做增量更新（分区、副本数）
        return TopicBuilder.name("my_topic1")//主题名称
                .partitions(3)//主题分区
                .replicas(3)//主题分区副本数
                .build();//创建
    }
}

3.6、屏蔽 kafka debug 日志 logback.xml

<configuration>　　　　　　
    <!-- 如果觉得idea控制台日志太多，src\main\resources目录下新建logback.xml
屏蔽kafka debug -->
    <logger name="org.apache.kafka.clients" level="debug" />
</configuration>

3.7、引入spring-kafka依赖

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>3.0.5</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>

    <!-- Generated by https://start.springboot.io -->
    <!-- 优质的 spring/boot/data/security/cloud 框架中文文档尽在 => https://springdoc.cn -->
    <groupId>com.atguigu.kafka</groupId>
    <artifactId>kafka-producer</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>kafka-producer</name>
    <description>kafka-producer</description>
    <properties>
        <java.version>17</java.version>
    </properties>
    <dependencies>
        <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.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

</project>

3.8、控制台日志

生产者即将发送消息：topic = my_topic1，partition:0，key = ，value = spring-kafka-事务1
生产者即将发送消息：topic = my_topic1，partition:0，key = ，value = spring-kafka-事务2
MyKafkaProducerListener消息发送失败：topic=my_topic1，partition = 0，key = ，value = spring-kafka-事务1，offset = -1
异常信息：Failing batch since transaction was aborted
MyKafkaProducerListener消息发送失败：topic=my_topic1，partition = 0，key = ，value = spring-kafka-事务2，offset = -1
异常信息：Failing batch since transaction was aborted

java.lang.ArithmeticException: / by zero

在这里插入图片描述

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mfbz.cn/a/693935.html

如若内容造成侵权/违法违规/事实不符，请联系我们进行投诉反馈qq邮箱809451989@qq.com，一经查实，立即删除！

相关文章

推荐云盘哪个好，各有各的优势

推荐云盘哪个好，各有各的优势

选择合适的云盘服务是确保数据安全、便捷分享和高效协作的关键。下面将从多个维度对目前主流的云盘服务进行详细的对比和分析&#xff1a; 速度性能 百度网盘青春版&#xff1a;根据测试&#xff0c;其上传和下载确实不限速&#xff0c;但主要定位是办公人群&#xff0c;适用于…
阅读更多...
JavaScript基础(十二)

JavaScript基础(十二)

截取字符串 //对象名.toLowerCase();将字符串转为小写 var strLAOWANG; strstr.toLowerCase(); console.log(str); //对象名.toUpperCase();将字符串转为大写 var str1laowang str1str1.toUpperCase(); console.log(str1); 截取字符串 //方法1&#xff1a;对象名.substr(a,b); …
阅读更多...
JS（JavaScript）的引用方式介绍与代码演示

JS（JavaScript）的引用方式介绍与代码演示

天行健&#xff0c;君子以自强不息&#xff1b;地势坤&#xff0c;君子以厚德载物。 每个人都有惰性&#xff0c;但不断学习是好好生活的根本&#xff0c;共勉&#xff01; 文章均为学习整理笔记&#xff0c;分享记录为主&#xff0c;如有错误请指正&#xff0c;共同学习进步。…
阅读更多...
android 抓取 logcat 日志的方法

android 抓取 logcat 日志的方法

1.找到这个路径 2.然后执行命令&#xff08;adb logcat -v time >.\\logcat.log&#xff09;&#xff0c;开始抓取日志 3.这个时候就可以去操作APP了&#xff0c;复现BUG了。 Ctrlc 结束日志抓取 adb logcat -c 清空旧日志
阅读更多...
seerfar选品功能，OZON运营插件工具seerfar

seerfar选品功能，OZON运营插件工具seerfar

在当今这个数字化、信息化的时代&#xff0c;电子商务的飞速发展使得越来越多的商家开始关注如何更高效地运营自己的在线店铺。其中&#xff0c;选品作为电商运营的重要一环&#xff0c;直接影响着店铺的流量、转化率和利润。在OZON这样的电商平台上&#xff0c;如何快速、准确…
阅读更多...
7天搞定Python必背500单词

7天搞定Python必背500单词

必备必记-你的Python就牛掰了 每天只背100个就足够了 老话说的好基础不扎实,地动山摇,在学习Python的时候前期基础很重要. 下面是大家常用遇到的Python基础单词,帮助你更好地掌握Python语言: 1.变量 在Python中用来存储数值,文本或其他信息的名称. 2. 函数 用于执行特定…
阅读更多...
力扣2444.统计定界子数组的数目

力扣2444.统计定界子数组的数目

力扣2444.统计定界子数组的数目 观察到不满足条件的数 可以作为天然的分割线 因此在枚举右端点的过程中 预处理minK&#xff0c;maxK和分割线上一次出现的下标 res min(min_i,max_i) - i0; 但是因为可能在到下个区段时 min_i和max_i尚未更新 导致结果为负数 所以要跟0再取一…
阅读更多...
linux：如何硬盘分区扩容

linux：如何硬盘分区扩容

文章目录 1. 前言2. 硬盘分区2.1 查看硬盘2.2 分区2.3 格式化 3. 硬盘分区扩容3.1 创建物理卷3.2 扩展到卷组&#xff08;volume group&#xff09;3.3 合并到待拓展分区3.4 使扩展生效 4 .参考 1. 前言 本文介绍如何将剩余的空间扩展到已有的硬盘分区中。 安装虚拟机的教程&…
阅读更多...
VBA高级应用30例应用2实现在列表框内及列表框间实现数据拖动

VBA高级应用30例应用2实现在列表框内及列表框间实现数据拖动

《VBA高级应用30例》&#xff08;版权10178985&#xff09;&#xff0c;是我推出的第十套教程&#xff0c;教程是专门针对高级学员在学习VBA过程中提高路途上的案例展开&#xff0c;这套教程案例与理论结合&#xff0c;紧贴“实战”&#xff0c;并做“战术总结”&#xff0c;以…
阅读更多...
离散数学答疑 4

离散数学答疑 4

知识点&#xff1a;什么是可结合&#xff1f; 举例A选项&#xff1a; 知识点&#xff1a;可交换性? 知识点&#xff1a;什么是阿贝尔群&#xff1f; 可交换->运算表中的元素关于主对角线对称 二阶子群的表达式 二阶子群作为一个群的子群&#xff0c;其本质是一个包含单位元…
阅读更多...
11 gpio 与 pinctrl 子系统

11 gpio 与 pinctrl 子系统

1、GPIO 硬件结构 GPIO 是通用输入/输出端口的简称。GPIO 的引脚与外部硬件设备连接，可实现与外部通讯、控制外部硬件或者采集外部硬件数据的功能。 八种工作模式 GPIO_Mode_AIN 模拟输入 GPIO_Mode_IN_FLOATING 浮空输入 GPIO_Mode_IPD 下拉输入 GPIO_Mode_IPU 上拉输入GP…
阅读更多...
深度学习中几种常见数据标准化方法

深度学习中几种常见数据标准化方法

目录 一、介绍 二、总结 三、详情 1. StandardScaler 2. MinMaxScaler 3. RobustScaler 4. MaxAbsScaler 5. Normalizer 6. QuantileTransformer 7. PowerTransformer 8. Log Transform 四、示例 五、心得 一、介绍 方法名称缩放范围适用条件StandardScaler均值…
阅读更多...
实用软件分享---简单菜谱 0.3版本 几千种美食（安卓）

实用软件分享---简单菜谱 0.3版本 几千种美食（安卓）

专栏介绍：本专栏主要分享一些实用的软件（Po Jie版）； 声明1：软件不保证时效性；只能保证在写本文时，该软件是可用的；不保证后续时间该软件能一直正常运行；不保证没有bug；如果软件不可用了，我知道后会第一时间在题目上注明（已失效）。介意者请勿订阅。 声明2：本专栏的…
阅读更多...
IO-源码阅读 glibc 2.35

IO-源码阅读 glibc 2.35

文章目录 参考缓存机制IO_FILE_PLUSfopenfopen_internal_IO_no_init_IO_old_init _IO_new_file_init_internal_IO_link_in _IO_new_file_fopen_IO_file_open fread_IO_fread_IO_sgetn_IO_doallocbuf_IO_file_doallocate_IO_file_stat_IO_setb __underflow_IO_new_file_underflo…
阅读更多...
深入Llama2：掌握未来语言模型的秘密

深入Llama2：掌握未来语言模型的秘密

Llama2是一个基于Transformer架构的大型语言模型&#xff0c;它旨在处理和理解大规模的文本数据。作为技术人员&#xff0c;了解Llama2的工作原理、模型结构和训练方法对于有效利用该模型至关重要。本文将详细介绍Llama2的基本概念、主要作用、使用方法及注意事项。 一、简介 …
阅读更多...
Linux---Linux编译器-gcc与g++的使用

Linux---Linux编译器-gcc与g++的使用

GCC是以GPL许可证所发行的自由软件&#xff0c;也是GNU计划的关键部分。GCC的初衷是为GNU操作系统专门编写一款编译器&#xff0c;现已被大多数类Unix操作系统&#xff08;如Linux、BSD、MacOS X等&#xff09;采纳为标准的编译器。 gcc是专门用来编译C语言的&#xff0c;而g是…
阅读更多...
VSC++： 民意调查比例法

VSC++： 民意调查比例法

void 民意调查比例法() {//缘由https://bbs.csdn.net/topics/396521294?page1#post-411408461从题目描述看&#xff1a;902/3~300.7&#xff0c;1498/5~299.6也就是大约求2个数的公约数&#xff0c;并使得这个公约数尽量求出最小误差&#xff1f;且商小于某值。int a 0, aa …
阅读更多...
Django Forbidden (CSRF cookie not set.)解决办法

Django Forbidden (CSRF cookie not set.)解决办法

解决办法就是在setting.py文件中注释&#xff1a; django.middleware.csrf.CsrfViewMiddleware, 这个中间件是为了防止跨站请求伪造的&#xff0c;平时用网页表单请求时&#xff0c;post提交是没有问题的&#xff0c;但是用api调用时就会被禁止&#xff0c;为了能使用接口调用…
阅读更多...
【Java面试】十七、并发篇（上）

【Java面试】十七、并发篇（上）

文章目录 1、synchronized关键字的底层原理&#xff1a;Monitor2、synchronized相关2.1 为什么说synchronized是重量级锁2.2 synchronized锁升级之偏向锁2.3 synchronized锁升级之轻量级锁 3、Java内存模型JMM4、CAS4.1 CAS流程4.2 CAS底层实现 5、volatile关键字的理解5.1 可见…
阅读更多...
使用python绘制季节图

使用python绘制季节图

使用python绘制季节图 季节图效果代码 季节图 季节图&#xff08;Seasonal Plot&#xff09;是一种数据可视化图表&#xff0c;用于展示时间序列数据的季节性变化。它通过将每个时间段&#xff08;如每个月、每个季度&#xff09;的数据绘制在同一张图表上&#xff0c;使得不同…
阅读更多...
最新文章

Copyright @ 2022~2023