0
0

Kafka入门使用教程

1. 前言

1.1. 什么是消息队列

消息队列（MQ）是消息传递中间件解决方案的一个组件，旨在支持独立的应用和服务之间的信息交换。 消息队列按发送顺序存储“消息”（由应用所创建、供其他应用使用的数据包），直到使用方应用能够处理它们为止。 这些消息安全地等待接收方应用做好准备，因此，即使网络或接收方应用出现问题，消息队列中的消息也不会丢失。

1.2. 为什么用消息队列

1.2.1. 解耦

生产者（客户端）发送消息到MQ中去，接受者（服务端）处理消息，需要消费的系统直接去MQ取消息进行消费即可而不需要和其他系统有耦合。

例如订单服务在电商订单创建后，发送扣减可用库存消息到MQ，库存系统在接受到消息后处理扣减可用库存。后续如果还需要扩展，通知其他第三方系统，只需要新增一个消费者去接收扣减信息处理即可。

1.2.2. 异步

将用户的请求数据存储到MQ之后就立即返回结果。随后，系统再对消息进行消费。

例如短信验证码业务，登录服务在接受到用户手机号之后，将手机号发送到MQ，短信服务接受消息后进行短信发送。登录服务不需要等待发送验证码完成，即可继续后续处理。

1.2.3. 削峰

先将短时间高并发产生的事务消息存储在MQ中，然后后端服务再慢慢根据自己的能力去消费这些消息。

例如电商大促时，可在订单服务前架设一层订单接收服务，只负责接收订单并将订单信息发送到MQ，订单服务根据自身消费能力来接收订单信息并创建订单。 这样就避免订单服务因过高流量而宕机。

1.3. Kafka

Kafka是最初由Linkedin公司开发，是一个分布式、支持分区的（partition）、多副本的（replica），基于zookeeper协调的分布式消息系统，它的最大的特性就是可以实时的处理大量数据以满足各种需求场景：比如基于hadoop的批处理系统、低延迟的实时系统、storm/Spark流式处理引擎，web/nginx日志、访问日志，消息服务等等，用scala语言编写，Linkedin于2010年贡献给了Apache基金会并成为顶级开源项目。

2. 安装

2.1. 官网

Apache Kafka

2.2. docker安装

  1. version: "1"
  2. services:
  3.   kafka:
  4.     image: 'bitnami/kafka:latest'
  5.     hostname: kafka
  6.     ports:
  7.       - 9092:9092
  8.       - 9093:9093
  9.     volumes:
  10.       - 'D:\Docker\Kafka\data:/bitnami/kafka'
  11.     networks:
  12.       - kafka_net
  13.     environment:
  14.       # KRaft settings
  15.       - KAFKA_CFG_NODE_ID=0
  16.       - KAFKA_CFG_PROCESS_ROLES=controller,broker
  17.       - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@kafka:9093
  18.       # Listeners
  19.       - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093
  20.       - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://192.168.2.51:9092
  21.       - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
  22.       - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
  23.       - KAFKA_CFG_INTER_BROKER_LISTENER_NAME=PLAINTEXT
  24. networks:
  25.   kafka_net:
  26.     driver: bridge
  1. docker-compose -f .\docker-compose.yml up -d

3. 简单使用

3.1. 可视化工具

3.1.1. offset explorer

Offset Explorer

  • 配置Cluster name名字,Bootrap servers地址等信息

  • 可以查看kafka节点，topic主题和consumers消费者等信息。

3.1.2. 可视化工具添加消息
  • 创建topic

  • 修改内容类型为String（记得点Update）

  • 选中Partition，点击加号，选择Add Single Message 。Key和Value都Enter Manualy，Key可不填。点击Add

  • 点击绿色Retrieve按钮，刷新分区消息。

3.2. 简单应用

3.2.1. 发送消息
  • 项目中添加Kafka依赖
  1. <!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients -->
  2. <dependency>
  3.   <groupId>org.apache.kafka</groupId>
  4.   <artifactId>kafka-clients</artifactId>
  5.   <version>3.7.0</version>
  6. </dependency>
  • 构建生产者者，填写对应的配置信息，模拟发送五次消息，并打印发送的信息
  1. Properties properties = new Properties();
  2. //修改为对应的bootstrap.servers
  3. properties.put("bootstrap.servers", "192.168.2.51:9092");
  4. properties.put("acks", "all");
  5. properties.put("retries", 0);
  6. properties.put("batch.size", 16384);
  7. properties.put("linger.ms", 1);
  8. properties.put("buffer.memory", 33554432);
  10. //序列化类型。
  11. properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
  12. properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
  13. Producer<String, String> producer = null;
  14. try {
  15.     producer = new KafkaProducer<>(properties);
  16.     for (int i = 0; i < 5; i++) {
  17.         String topic = "my-topic";
  18.         String value = "My first message" + i;
  19.         ProducerRecord<String, String> producerRecord = new ProducerRecord<>(topic, value);
  20.         producer.send(producerRecord);
  21.         System.out.println("Sent:" + producerRecord);
  22.     }
  23. } catch (Exception e) {
  24.     e.printStackTrace();
  26. } finally {
  27.     producer.close();
  28. }
  • 效果展示

  • 工具查看Kafka中的消息

3.2.2. 消费消息
  • 构建消费者，填写对应的配置信息，while循环模拟消费线程，并打印消费的信息
  1. Properties properties = new Properties();
  2. //修改为对应的bootstrap.servers
  3. properties.put("bootstrap.servers", "192.168.2.51:9092");
  4. //可自行修改，消费组名称
  5. properties.put("group.id", "zzj.kafka");
  6. properties.put("enable.auto.commit", "true");
  7. properties.put("auto.commit.interval.ms", "1000");
  8. properties.put("auto.offset.reset", "earliest");
  9. properties.put("session.timeout.ms", "30000");
  10. //反序列化
  11. properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  12. properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  13. KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
  14. //制定消费的topic主题
  15. kafkaConsumer.subscribe(Arrays.asList("my-topic"));
  16. while (true) {
  17.     ConsumerRecords<String, String> records = kafkaConsumer.poll(100);//100是超时时间
  18.     for (ConsumerRecord<String, String> record : records) {
  19.         System.out.printf("offset = %d, value = %s", record.offset(), record.value());
  20.         System.out.println();
  21.     }
  22. }
  • 效果展示

  • 工具查看消费记录，选择对应的group.id消费组。可查看offset偏移量等信息

3.3. Springboot项目应用

3.3.1. Springboot集成Kafka
  • 项目中添加Kafka依赖
  1. <dependency>
  2.   <groupId>org.springframework.kafka</groupId>
  3.   <artifactId>spring-kafka</artifactId>
  4. </dependency>
  • 添加Kafka生成者和消费者配置
  1. spring.kafka.bootstrap-servers=localhost:9092
  2. spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
  3. spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
  • 模拟生产者代码，使用spring-kafak封装的KafkaTemplate
  1. import lombok.extern.slf4j.Slf4j;
  2. import org.springframework.beans.factory.annotation.Autowired;
  3. import org.springframework.http.HttpStatus;
  4. import org.springframework.http.ResponseEntity;
  5. import org.springframework.kafka.core.KafkaTemplate;
  6. import org.springframework.web.bind.annotation.*;
  8. @RestController
  9. @RequestMapping("kafka")
  10. @Slf4j
  11. public class KafkaController {
  13.     @Autowired
  14.     private KafkaTemplate<String, String> kafkaTemplate;
  16.     @RequestMapping(method = RequestMethod.POST, value = "/send")
  17.     public ResponseEntity<String> send(@RequestBody String body){
  18.         log.info("send message: {}", body);
  19.         kafkaTemplate.send("my-springboot-topic", body);
  20.         return new ResponseEntity<>("success", HttpStatus.OK);
  21.     }
  22. }
  • 消费者代码，使用@KafkaListener注解配置消费方法
  1. package com.zzj.kafkademo.demos;
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.springframework.kafka.annotation.KafkaListener;
  5. import org.springframework.kafka.annotation.TopicPartition;
  6. import org.springframework.kafka.support.KafkaHeaders;
  7. import org.springframework.messaging.handler.annotation.Header;
  8. import org.springframework.messaging.handler.annotation.Payload;
  9. import org.springframework.stereotype.Component;
  11. @Component
  12. @Slf4j
  13. public class KafkaConsumer {
  15.     @KafkaListener(topics = {"my-springboot-topic"}, groupId = "springboot-consumer")
  16.     public void listen(
  17.             //接收消息体
  18.             @Payload String data,
  20.             //接收分区编号
  21.             @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
  23.             //接收topic主题名
  24.             @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
  26.             //接收消息的时间戳
  27.             @Header(KafkaHeaders.RECEIVED_TIMESTAMP) long ts
  28.     ){
  29.         log.info("data:{},partition:{},topic:{},ts:{}", data, partition, topic, ts);
  30.     }
  31. }
3.3.2. 效果展示
  • 使用postman模拟发送消息

  • 查看发送和消费记录，KafkaController中打印发送日志，KafkaConsumer打印消费日记

4. 问题总结

4.1. 系统可用性降低：

系统可用性在某种程度上说是降低了的，为什么这样说呢？在引入Kafka之后，就需要考虑Kafka的稳定性，如果突然挂了要如何处理，如何避免Kafka挂掉等问题。

4.2. 系统复杂性提高：

引入Kafka之后，需要保证消息没有被重复消费、消息是否丢失、还要保证消息传递的顺序性，消息消费错误导致的数据一致性问题等情况。

标签： kafka linq 分布式
本文转载自: https://blog.csdn.net/qq_42322158/article/details/138142202
版权归原作者 梦里的银渐层 所有， 如有侵权，请联系我们删除。
登录后发布评论

“Kafka入门使用教程”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？

文章导航
没有找到文章内容分段。。。