kafka消息队列的初步探索

news/2024/10/30 13:34:31/

消息队列的作用就是提高运行速度,防止线程堵塞。

kafka的作用

    异步

 通过在消息队列发送消息的方式,将对应的业务作为监听者,此时我们只需要考虑发送消息的时间即可,大大提高了运行的速度。

    解耦

 

如果使用原来的直接调用对应业务的方式,在被调用业务发生修改是,调用业务也需要修改代码,存在很大的耦合,所以使用消息队列的方式,后续我们只需要关注消息的发送,无需关注业务的内部实现,大大的降低了耦合性。 

     削峰

在一些业务场景小(如:限时秒杀),此时在同一个时间内会有大量的请求发向服务器,这就会导致服务器瘫痪,所以这里引入的消息队列的方式,这些请求会一一的给消息队列发送消息,服务器通过一次处理对应个数的消息来处理对应的请求,最终实现削峰,防止服务器瘫痪。

     缓冲

 和削峰类似就是通过消息队列的形式处理请求,防止服务器瘫痪。

消息模式 

1.消息点对点模式

 一对一的形式,消费者每次从消息队列中接收一个消息,在确定接收后,消息队列就会将刚刚被接收的消息从消息队列中删除。

2.消息发布订阅模式

 在消息队列中存储的消息会被分为不同的主题里(其实就是将这些消息进行分组), 消费者就是去订阅对应的topic,消费者也可以组成对应的消费者组,此时消费者就从对应的topic中获取对应的消息,在其中存在偏移量这个数据(offset),通过该偏移量获取对应的位置的消息。重点来了,在该消息队列中的消息在被使用后是不会被移除的

kafka工作原理 

 在未来的项目中,我们大多都是已微服务的形式进行开发,此时消息队列中同个topic中的消息可能会存在于不用的服务器上,这就是进行分区。为了防止其中某太服务器发生宕机后影响项目的运行,我们可以在对应分区中存储其他分区中的消息,实现备份在宕机时不影响项目的运行,此过程就是创建副本

 springboot整合kafka

导入kafka整合springboot依赖

<dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency><dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency>

 发布消息

import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.core.KafkaTemplate;@SpringBootTest
class KafkatestApplicationTests {@AutowiredKafkaTemplate kafkaTemplate;@Testvoid test1() {
//设置默认的主题kafkaTemplate.setDefaultTopic("tiktop");
//在对应主题中添加消息,此消息以键值对的形式kafkaTemplate.send("tiktop", "抖音消息", "你好,秃狼");}}

 测试结果为下:

特殊情况(无法识别到主机)

 解决方法(在hosts中设置主机地址)

 通过火绒修改hosts。

 设置消息的value为实体类类型

我们通过application.properties进行设置。

#设置消息值的类型,这里设置为json类型,这样我们就可以在消息中传入实体类
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer

默认的value的序列化类型是: 

可以设置的序列化类型为下: 

 测试结果为下:

创建消费者 

 创建消费者

group表示该消费者的分组。topicPartitions是监听的所有的topic和分区,@TopicPartition就是设置对应topic区的topic名字,和对应偏移量和分区(在监听中可能会同时监听多个topic)。partitionOffsets就是设置所有分区和偏移量(在监听中可能同时监听多个分区中,在该分区中会有不同的偏移量)
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.PartitionOffset;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.stereotype.Component;@Component
public class Listener {/*** group表示该消费者的分组* topicPartitions是监听的所有的topic和分区,@TopicPartition就是设置对应topic区的topic名字,和对应偏移量和分区(在监听中可能会同时监听多个topic)* partitionOffsets就是设置所有分区和偏移量(在监听中可能同时监听多个分区中,在该分区中会有不同的偏移量)*/@KafkaListener(groupId = "toktop-server", topicPartitions = {@TopicPartition(topic = "tiktop", partitionOffsets = {@PartitionOffset(partition = "0", initialOffset = "0")})})public void listen(ConsumerRecord consumerRecord) {//ConsumerRecord就是整个消费者的信息Object key = consumerRecord.key();System.out.println("key=" + key);Object value = consumerRecord.value();System.out.println("value=" + value);}
}

在启动类上添加kafka的注解驱动,这样@KafkaListener才会被识别。(@Enablekafka)

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.kafka.annotation.EnableKafka;@SpringBootApplication
//开启kafka的注解驱动
@EnableKafka
public class KafkatestApplication {public static void main(String[] args) {SpringApplication.run(KafkatestApplication.class, args);}}

进行启动测试,测试结果为下

 在启动后消费者会直接监听消息队列,测试我们将偏移量设置为0,也就是从头部开始,此时消费者监听到消息队列中的两个消息,最终将通过的信息输出。(注意使用的模式是:发布和订阅模式,所以接收到消息后不会将消息删除,而是改变偏移量

kafka自动配置

kafka 自动配置在KafkaAutoConfiguration

  1. 容器中放了 KafkaTemplate 可以进行消息收发
  2. 容器中放了KafkaAdmin 可以进行 Kafka 的管理,比如创建 topic 等
  3. kafka 的配置在KafkaProperties中
  4. @EnableKafka可以开启基于注解的模式


http://www.ppmy.cn/news/516247.html

相关文章

Could not autowire. No beans of ‘xx‘ type found异常

目录 可能没有将所需的类声明为一个Bean&#xff1a; 导包导错了&#xff0c;这个就不多说还不赶紧看看自己的包是不是导错了&#xff1b; 配置类可能没有全部加载 XML中忘记生成bean&#xff1a; idea工具的显示报错&#xff1a;(一般不会是这个问题) 可能没有将所需的类…

【Red Hat 7.9---详细安装Oracle 11g】---图形化界面方式

【Red Hat 7.9---详细安装Oracle 11g】---图形化界面方式 &#x1f53b; 一、安装前规划&#x1f53b; 二、安装前准备一&#xff08;系统参数修改&#xff09;⛳ 2.1 内核版本、系统版本查看⛳ 2.2 修改主机名-重启生效⛳ 2.3 关闭selinux⛳ 2.4 防火墙设置1521端口开放⛳ 2.5…

Linux的基础操作

0.前言 您好&#xff0c;这里是limou3434的一篇关于Linux基础操作的博文。感兴趣的话&#xff0c;可以看看我的其他博文系列。本次我给您带来的是Linux下的几个基础命令&#xff0c;学会这些命令您就可以开始在Linux管理和操作一些文件了。 注意&#xff1a;本次演示环境是在…

机械革命极光air和pro的区别 选哪个好

机械革命极光air&#xff1a;其平时的售价行情是在6999元左右 机械革命极光pro&#xff1a;其电商的平时售价行情是在8699元左右 机械革命极光air的主要优势是在机身的轻薄方面&#xff0c;更方便外出携带&#xff0c;而机械革命极光pro在显卡、屏幕素质、电池容量方面更有优势…

关于机械革命电脑关机后自动重启的解决方案

本人的机械革命CODE01&#xff0c;之前因为需要写代码&#xff0c;所以长时间保持开机状态没管&#xff0c; 最近需要关机带走的时候发现它关不了机&#xff0c;电源灯一直亮着&#xff0c;而且关机之后会自动重启 原因嘛&#xff0c;其实很简单&#xff0c;就是BIOS设置出了问…

中国移动H1S-3光猫首发破解路由器桥接教程

进入正题&#xff0c;开始破解 1&#xff0e; 打开Telnet 使用工具&#xff1a;Firefox浏览器和HTTPHeader Live插件 使用光猫后面的普通账号&#xff08;user&#xff09;登陆&#xff0c;打开HTTP Header Live插件&#xff0c;在页面上随便点击一个页面&#xff0c;在看HTTP …

机械革命Z2air实现 Wake On Lan 远程开机

背景 几个月前&#xff0c;我就想要实现远程开机笔记本。配合上RD Client实现真正意义上的远程。 但是当时遇到了各种问题&#xff0c;首先是进不了BIOS&#xff0c;还一度卡开机logo。后来查明原来是硬盘的原因&#xff0c;便慷慨地把那块512G的SSD送给了wy同志2333。wy同志…

机械革命无线网消失解决办法

机械革命无线网突然显示不出来 1、首先winR 2、输入regedit 3、找到计算机目录下的 \HKEY_LOCAL_MACHINE\SYSTEM\CurrentControlSet\Services\Ndisuio 4、找到DisplayName数据值右键看看数值数据 NDIS Usermode I/O Protocol 5、找到Start数据值右键看看数值数据&#xf…