kafka消息队列的初步探索

news2024/9/21 3:23:30

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

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 {

    @Autowired
    KafkaTemplate kafkaTemplate;

    @Test
    void 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);
    }

}

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