kafka微服务学习

消息中间件对比：
1、吞吐、可靠性、性能
在这里插入图片描述

Kafka安装

Kafka对于zookeeper是强依赖，保存kafka相关的节点数据，所以安装Kafka之前必须先安装zookeeper

  • Docker安装zookeeper

下载镜像：

docker pull zookeeper:3.4.14

创建容器

docker run -d --name zookeeper -p 2181:2181 zookeeper:3.4.14
  • Docker安装kafka

下载镜像：

docker pull wurstmeister/kafka:2.12-2.3.1

创建容器

docker run -d --name kafka \
--env KAFKA_ADVERTISED_HOST_NAME=192.168.200.130 \
--env KAFKA_ZOOKEEPER_CONNECT=192.168.200.130:2181 \
--env KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.200.130:9092 \
--env KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 \
--env KAFKA_HEAP_OPTS="-Xmx256M -Xms256M" \
--net=host wurstmeister/kafka:2.12-2.3.1

kafka入门

  • 生产者发送消息，多个消费者只能有一个消费者接收到消息
  • 生产者发送消息，多个消费者都可以接收到消息

（1）创建kafka-demo项目，导入依赖

<dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId>
</dependency>

（2）生产者发送消息

package com.heima.kafka.sample;import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;import java.util.Properties;/*** 生产者*/
public class ProducerQuickStart {public static void main(String[] args) {//1.kafka的配置信息Properties properties = new Properties();//kafka的连接地址properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.200.130:9092");//发送失败，失败的重试次数properties.put(ProducerConfig.RETRIES_CONFIG,5);//消息key的序列化器properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");//消息value的序列化器properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");//2.生产者对象KafkaProducer<String,String> producer = new KafkaProducer<String, String>(properties);//封装发送的消息ProducerRecord<String,String> record = new ProducerRecord<String, String>("itheima-topic","100001","hello kafka");//3.发送消息producer.send(record);//4.关闭消息通道，必须关闭，否则消息发送不成功producer.close();}}

（3）消费者接收消息

package com.heima.kafka.sample;import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;import java.time.Duration;
import java.util.Collections;
import java.util.Properties;/*** 消费者*/
public class ConsumerQuickStart {public static void main(String[] args) {//1.添加kafka的配置信息Properties properties = new Properties();//kafka的连接地址properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.200.130:9092");//消费者组properties.put(ConsumerConfig.GROUP_ID_CONFIG, "group2");//消息的反序列化器properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");//2.消费者对象KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);//3.订阅主题consumer.subscribe(Collections.singletonList("itheima-topic"));//当前线程一直处于监听状态while (true) {//4.获取消息ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofMillis(1000));for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {System.out.println(consumerRecord.key());System.out.println(consumerRecord.value());}}}}

kafka高可用设计

1、设计集群模式：

Kafka的服务器端由被称为 Broker 的服务进程构成，即一个 Kafka 集群由多个Broker 组成。当一个机器宕机了，另外一个机器就会替补山
在这里插入图片描述

2、备份机制：

Kafka定义了两类副本

  1. 领导者副本(Leader Replica)
  2. 追随者副本 (Follower Replica)
    追随者副本分为两类：
    1、一种是ISR副本，同步保存
    2、普通的副本，异步保存
    出现主节点宕机，会先选ISR副本中的一个成为新的主节点，保证数据一致性，没有ISR节点，再从普通节点中挑选
    针对全部节点宕机的情况，有两种策略：
    1、等待第一个ISR副本，保证了数据的尽可能一致
    2、等待一个复活的追随者，无论是ISR还是普通，提高系统的高可用性。

kafka生产者详解

1发送类型

  • 同步发送

    使用send()方法发送，它会返回一个Future对象，调用get()方法进行等待，就可以知道消息是否发送成功

RecordMetadata recordMetadata = producer.send(kvProducerRecord).get();
System.out.println(recordMetadata.offset());

  • 异步发送

    调用send()方法，并指定一个回调函数，服务器在返回响应时调用函数

//异步消息发送
producer.send(kvProducerRecord, new Callback() {@Overridepublic void onCompletion(RecordMetadata recordMetadata, Exception e) {if(e != null){System.out.println("记录异常信息到日志表中");}System.out.println(recordMetadata.offset());}
});

2参数详解

  • ack

代码的配置方式：

//ack配置  消息确认机制
prop.put(ProducerConfig.ACKS_CONFIG,"all");

参数的选择说明

确认机制说明
acks=0生产者在成功写入消息之前不会等待任何来自服务器的响应,消息有丢失的风险，但是速度最快
acks=1（默认值）只要集群首领节点收到消息，生产者就会收到一个来自服务器的成功响应
acks=all只有当所有参与赋值的节点全部收到消息时，生产者才会收到一个来自服务器的成功响应
  • retries

生产者从服务器收到的错误有可能是临时性错误，在这种情况下，retries参数的值决定了生产者可以重发消息的次数，如果达到这个次数，生产者会放弃重试返回错误，默认情况下，生产者会在每次重试之间等待100ms

代码中配置方式：

//重试次数
prop.put(ProducerConfig.RETRIES_CONFIG,10);
  • 消息压缩

默认情况下， 消息发送时不会被压缩。

代码中配置方式：

//数据压缩
prop.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"lz4");
压缩算法说明
snappy占用较少的 CPU， 却能提供较好的性能和相当可观的压缩比， 如果看重性能和网络带宽，建议采用
lz4占用较少的 CPU， 压缩和解压缩速度较快，压缩比也很客观
gzip占用较多的 CPU，但会提供更高的压缩比，网络带宽有限，可以使用这种算法

使用压缩可以降低网络传输开销和存储开销，而这往往是向 Kafka 发送消息的瓶颈所在。

kafka消费者

消息的有序性

方法：一个topic分区能保证自己的数据是按照先后消费的，但是不能保证跨分区消息处理的先后顺序。我么只能使用一个分区，在单分区种，消息可以保证严格顺序消费

提交和偏移量

在这里插入图片描述
自动提交：
当enable.auto.commit被设置为true，提交方式就是让消费者自动提交偏移量，每隔5秒消费者会自动把从poll0方法接收的最大偏移量提交上去，这样只是记录了规定时间内的最大偏移量，其实与数据提交的偏移量存在偏差，因此可能会出现数据的重复提交或者丢失
手动提交
当enableauto.commit被设置为false可以有以下三种提交方式

  • 提交当前偏移量(同步提交)
  • 异步提交
  • 同步和异步组合提交

同步提交：commitSync()方法会一直尝试直至提交成功，如果提交失败也可以记录到错误日志里。

while (true){ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));for (ConsumerRecord<String, String> record : records) {System.out.println(record.value());System.out.println(record.key());try {consumer.commitSync();//同步提交当前最新的偏移量}catch (CommitFailedException e){System.out.println("记录提交失败的异常："+e);}}
}

异步提交：手动提交有一个缺点，那就是当发起提交调用时应用会阻塞。消息没有重试机制

while (true){ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));for (ConsumerRecord<String, String> record : records) {System.out.println(record.value());System.out.println(record.key());}consumer.commitAsync(new OffsetCommitCallback() {@Overridepublic void onComplete(Map<TopicPartition, OffsetAndMetadata> map, Exception e) {if(e!=null){System.out.println("记录错误的提交偏移量："+ map+",异常信息"+e);}}});
}

同步和异步组合提交

异步提交也有个缺点，那就是如果服务器返回提交失败，异步提交不会进行重试。相比较起来，同步提交会进行重试直到成功或者最后抛出异常给应用。异步提交没有实现重试是因为，如果同时存在多个异步提交，进行重试可能会导致位移覆盖。

举个例子，假如我们发起了一个异步提交commitA，此时的提交位移为2000，随后又发起了一个异步提交commitB且位移为3000；commitA提交失败但commitB提交成功，此时commitA进行重试并成功的话，会将实际上将已经提交的位移从3000回滚到2000，导致消息重复消费。

try {while (true){ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));for (ConsumerRecord<String, String> record : records) {System.out.println(record.value());System.out.println(record.key());}consumer.commitAsync();}
}catch (Exception e){+e.printStackTrace();System.out.println("记录错误信息："+e);
}finally {try {consumer.commitSync();}finally {consumer.close();}
}

springboot整合kafka

1、在父类中的pop文件中导入依赖包

```xml
<!-- kafkfa -->
<dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId>
</dependency>
<dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId>
</dependency>

2、在需要用到kafka的微服务的naco中分别配置生产者和消费者配置

spring:kafka:bootstrap-servers: 192.168.200.130:9092producer:retries: 10key-serializer: org.apache.kafka.common.serialization.StringSerializervalue-serializer: org.apache.kafka.common.serialization.StringSerializer

spring:kafka:bootstrap-servers: 192.168.200.130:9092consumer:group-id: ${spring.application.name}key-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.apache.kafka.common.serialization.StringDeserializer

传递消息为对象

目前springboot整合后的kafka，因为序列化器是StringSerializer，这个时候如果需要传递对象可以有两种方式

方式一：可以自定义序列化器，对象类型众多，这种方式通用性不强，本章节不介绍

方式二：可以把要传递的对象进行转json字符串，接收消息后再转为对象即可，本项目采用这种方式

  • 发送消息
@GetMapping("/hello")
public String hello(){User user = new User();user.setUsername("xiaowang");user.setAge(18);kafkaTemplate.send("user-topic", JSON.toJSONString(user));return "ok";
}
  • 接收消息
package com.heima.kafka.listener;import com.alibaba.fastjson.JSON;
import com.heima.kafka.pojo.User;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;@Component
public class HelloListener {@KafkaListener(topics = "user-topic")public void onMessage(String message){if(!StringUtils.isEmpty(message)){User user = JSON.parseObject(message, User.class);System.out.println(user);}}
}

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

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

CHM Viewer Star 6.3.2(CHM文件阅读)

CHM Viewer Star 6.3.2(CHM文件阅读)

CHM Viewer Star 是一款适用于 Mac 平台的 CHM 文件阅读器软件&#xff0c;支持本地和远程 CHM 文件的打开和查看。它提供了直观易用的界面设计&#xff0c;支持多种浏览模式&#xff0c;如书籍模式、缩略图模式和文本模式等&#xff0c;并提供了丰富的功能和工具&#xff0c;如…
阅读更多...
02MyBatisPlus条件构造器，自定义SQL，Service接口

02MyBatisPlus条件构造器，自定义SQL，Service接口

一、条件构造器 1.MyBatis支持各种复杂的where条件&#xff0c;满足开发的需求 Wrapper是条件构造器&#xff0c;构建复杂的where查询 AbstractWrapper有构造where条件的所有方法&#xff0c;QueryWrapper继承后并有自己的select指定查询字段。UpdateWrapper有指定更新的字段的…
阅读更多...
Ocelot：.NET开源API网关提供路由管理、服务发现、鉴权限流等功能

Ocelot：.NET开源API网关提供路由管理、服务发现、鉴权限流等功能

随着微服务的兴起&#xff0c;API网关越来越常见。API网关是连接应用程序和用户之间的桥梁&#xff0c;就像一个交通指挥员&#xff0c;负责处理所有进出应用的数据和请求&#xff0c;确保安全、高效、有序地流通。 今天给大家推荐一个.NET开源API网关。 01 项目简介 Ocelot…
阅读更多...
C# ZXing 二维码，条形码生成与识别

C# ZXing 二维码，条形码生成与识别

C# ZXing 二维码条形码生成识别 安装ZXing使用ZXing生成条形码生成二维码生成带Logo的二维码识别二维码、条形码 安装ZXing NuGet搜索ZXing安装ZXing.Net包 使用ZXing using ZXing; using ZXing.Common; using ZXing.QrCode; using ZXing.QrCode.Internal; 生成条形码 //…
阅读更多...
【算法】牛的旅行（图的直径，floyd算法求最短路）

【算法】牛的旅行（图的直径，floyd算法求最短路）

题目 农民John的农场里有很多牧区&#xff0c;有的路径连接一些特定的牧区。 一片所有连通的牧区称为一个牧场。 但是就目前而言&#xff0c;你能看到至少有两个牧区不连通。 现在&#xff0c;John想在农场里添加一条路径&#xff08;注意&#xff0c;恰好一条&#xff09;。 一…
阅读更多...
基于Python+OpenCV+SVM车牌识别系统-车牌预处理系统

基于Python+OpenCV+SVM车牌识别系统-车牌预处理系统

欢迎大家点赞、收藏、关注、评论啦 &#xff0c;由于篇幅有限&#xff0c;只展示了部分核心代码。 文章目录 一项目简介简介系统流程系统优势 二、功能三、系统四. 总结 一项目简介 ## PythonOpenCVSVM车牌识别系统介绍 简介 PythonOpenCVSVM车牌识别系统是一种基于计算机视…
阅读更多...
【小沐学写作】PPT、PDF文件添加水印（Python）

【小沐学写作】PPT、PDF文件添加水印（Python）

文章目录 1、简介2、ppt添加水印2.1 PowerPoint幻灯片母版2.2 iSlide插件&#xff08;收费&#xff09;2.2.1 iSlide简介2.2.2 iSlide定价2.2.3 iSlide水印 2.3 Python代码2.3.1 Aspose.Slides for Python&#xff08;收费&#xff09; 3、pdf添加水印3.1 Python代码3.1.1 PyPD…
阅读更多...
Longhorn跨AZ实现存储高可用

Longhorn跨AZ实现存储高可用

Longhorn跨AZ实现存储高可用 longhorn基础组件功能及其作用这里就不做介绍了 方案一 Longhorn跨AZ的高可用的就是一个PVC的replicas 均匀打散的不同的AZ区域之间&#xff0c;这样当某个AZ挂掉后&#xff0c;engine会立即使用另外一个数据副本&#xff0c;并重建这个副本&…
阅读更多...
《红蓝攻防对抗实战》十一.内网穿透之利用SSH协议进行隧道穿透

《红蓝攻防对抗实战》十一.内网穿透之利用SSH协议进行隧道穿透

利用DNS协议进行隧道穿透 一.前言二.前文推荐三. 利用SSH协议进行隧道穿透1.SSH隧道-本地端口转发2.SSH隧道-远程端口转发3.SSH隧道-动态端口转发 四.本篇总结 一.前言 SSH&#xff08;Secure Shell&#xff09;协议是一种加密的网络传输协议&#xff0c;它可以在不安全的网络…
阅读更多...
在CMake中打印日志信息

在CMake中打印日志信息

message([STATUS|WARNING|AUTHOR_WARNING|FATAL_ERROR|SEND_ERROR] "message to display" ...) (无) &#xff1a;重要消息 STATUS &#xff1a;非重要消息 WARNING&#xff1a;CMake 警告, 会继续执行 AUTHOR_WARNING&#xff1a;CMake 警告 (dev), 会继续执行 SEN…
阅读更多...
代码随想录 Day43 动态规划11 LeetCode T309 买卖股票的最佳时期含冷冻期 T714买卖股票的最佳时机含手续费

代码随想录 Day43 动态规划11 LeetCode T309 买卖股票的最佳时期含冷冻期 T714买卖股票的最佳时机含手续费

LeetCode T309 买卖股票的最佳时机含冷冻期 题目链接:309. 买卖股票的最佳时机含冷冻期 - 力扣&#xff08;LeetCode&#xff09; 题目思路: 这题其实就是将卖出的状态拆分成三个状态 1.前两天就卖出并一直保持卖出的状态 2.今天卖出的状态 3.今天是冷冻期的状态 当然还有一个…
阅读更多...
一个java文件的JVM之旅

一个java文件的JVM之旅

准备 我是小C同学编写得一个java文件&#xff0c;如何实现我的功能呢&#xff1f;需要去JVM(Java Virtual Machine)这个地方旅行。 变身 我高高兴兴的来到JVM&#xff0c;想要开始JVM之旅&#xff0c;它确说&#xff1a;“现在的我还不能进去&#xff0c;需要做一次转换&#x…
阅读更多...
【左程云算法全讲7】二叉树基础

【左程云算法全讲7】二叉树基础

系列综述&#xff1a; &#x1f49e;目的&#xff1a;本系列是个人整理为了秋招面试的&#xff0c;整理期间苛求每个知识点&#xff0c;平衡理解简易度与深入程度。 &#x1f970;来源&#xff1a;材料主要源于左程云算法课程进行的&#xff0c;每个知识点的修正和深入主要参考…
阅读更多...
数据库数据恢复—无备份，未开启binlog的MySQL误删除怎么恢复数据

数据库数据恢复—无备份，未开启binlog的MySQL误删除怎么恢复数据

数据库数据恢复环境&#xff1a; 一台本地windows sever操作系统服务器&#xff0c;服务器上部署mysql数据库单实例&#xff0c;引擎类型为innodb&#xff0c;表内数据存储所使用表空间类型为独立表空间。无数据库备份&#xff0c;未开启binlog。 数据库故障&分析&#xf…
阅读更多...
玩了个锤子游戏小程序搭建流程：探索深度与逻辑的结合

玩了个锤子游戏小程序搭建流程：探索深度与逻辑的结合

随着移动互联网的普及&#xff0c;小程序已经成为了越来越多用户的选择。在这个背景下&#xff0c;玩了个锤子游戏小程序应运而生&#xff0c;它为用户提供了一个全新的游戏体验。那么&#xff0c;如何搭建这样一个小程序呢&#xff1f;本文将为大家详细介绍玩了个锤子游戏小程…
阅读更多...
一文图解爬虫（spider）

一文图解爬虫（spider）

—引导语 互联网&#xff08;Internet&#xff09;进化到今天&#xff0c;已然成为爬虫&#xff08;Spider&#xff09;编制的天下。从个体升级为组合、从组合联结为网络。因为有爬虫&#xff0c;我们可以更迅速地触达新鲜“网事”。 那么爬虫究竟如何工作的呢&#xff1f;允许…
阅读更多...
守护进程daemon()，C 库函数asctime、localtime，UDEV的配置文件，开机自启动，自动挂载U盘

守护进程daemon()，C 库函数asctime、localtime，UDEV的配置文件，开机自启动，自动挂载U盘

一、守护进程 二、daemon()函数 三、C 库函数asctime、localtime 四、设置守护进程开机自启动 五、守护进程应用 编写判断守护进程是否在运行的程序 守护进程不让控制程序退出 把相关守护进程设置成开机自启动 六、dmesg 七、UDEV的配置文件&#xff08;udev的rules编写&am…
阅读更多...
clang插件对llvm源码插桩，分析函数调用日志(1)--google镜像

clang插件对llvm源码插桩，分析函数调用日志(1)--google镜像

tick_plot__compile.ipynb 时长边界_时上链异数: 长短函数调用链列表 0. 用matplotlib找系统中字体文件大于1MB的 中文字体通常很大&#xff0c;这样过滤出的 通常有中文字体 结果中 看名字 ‘AR PL UMing CN’ 果然是中文字体 from matplotlib.font_manager import fontManag…
阅读更多...
景联文科技助力金融机构强化身份验证，提供高质量人像采集服务

景联文科技助力金融机构强化身份验证，提供高质量人像采集服务

随着社会的数字化和智能化进程的加速&#xff0c;人像采集在金融机构身份认证领域中发挥重要作用&#xff0c;为人们的生活带来更多便利和安全保障。 金融机构在身份验证上的痛点主要包括以下方面&#xff1a; 身份盗用和欺诈风险&#xff1a;传统身份验证方式可能存在漏洞&am…
阅读更多...
【已解决】ModuleNotFoundError: No module named ‘sklearn‘

【已解决】ModuleNotFoundError: No module named ‘sklearn‘

问题描述 Traceback (most recent call last): File "/home/visionx/nickle/temp/SimCLR/linear_evaluation.py", line 210, in <module> from sklearn.manifold import TSNE ModuleNotFoundError: No module named sklearn 解决办法 pip install numpy…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023