温岭市市住房和城乡建设规划局网站山东省建设从业人员管理系统入口

温岭市市住房和城乡建设规划局网站,山东省建设从业人员管理系统入口,沈阳网站制作企业,排名前50名免费的网站回顾完RabbitMQ#xff0c;再跟我一起回顾下Kafka ~一、Kafka介绍1. 什么是Kafka#xff1f;Kafka是由Apache软件基金会开发的分布式流处理平台#xff0c;最初由LinkedIn公司设计#xff0c;现已成为大数据领域核心的消息中间件。它能处理实时数据流#xff0c;支持高吞吐…回顾完RabbitMQ再跟我一起回顾下Kafka ~一、Kafka介绍1. 什么是KafkaKafka是由Apache软件基金会开发的分布式流处理平台最初由LinkedIn公司设计现已成为大数据领域核心的消息中间件。它能处理实时数据流支持高吞吐、低延迟、可扩展的消息传递广泛用于日志收集、实时分析、事件驱动架构等场景。2. 核心特点高吞吐单机可支持百万级消息/秒通过分区并行处理实现。持久化消息持久化到磁盘支持TB级数据存储默认保留7天。分布式集群由多个Broker服务器组成支持水平扩展。多订阅者一个Topic的消息可被多个消费者组独立消费广播/负载均衡。二、Kafka架构与核心组件1. 核心组件组件 作用Broker Kafka服务器节点存储Topic数据每个Broker有唯一IDbroker.id。Topic 消息的逻辑分类如order-topic类似“消息频道”包含多个Partition。Partition Topic的物理分片有序日志文件分布式存储的基本单位每个Partition有Leader和Follower副本。Producer 发送消息到Topic的客户端如订单服务。Consumer 从Topic订阅消息的客户端如库存服务。Consumer Group 消费者组组内多个消费者负载均衡消费Partition组间独立消费广播。2. 架构图MermaidKafka Cluster发送消息分区存储分区存储同步数据负载均衡消费协调协调协调管理消费者组Broker 1broker.id0• TopicA-Partition0 Leader• TopicB-Partition1 FollowerBroker 2broker.id1• TopicA-Partition1 Leader• TopicB-Partition0 LeaderBroker 3broker.id2• TopicA-Partition0 FollowerZooKeeper集群协调存储元数据Producer发送消息到TopicConsumer Group组内负载均衡消费TopicA• Partition0• Partition1三、消息流转完整路径生产者→消费者1. 流转步骤生产者发送消息生产者指定Topic和Key可选通过分区器将消息分配到Partition默认按Key哈希。Broker存储消息Leader副本接收消息并写入磁盘Segment文件Follower副本同步数据。消费者组分配Partition消费者组启动时协调者Coordinator将Topic的Partition分配给组内消费者一个Partition仅被一个消费者消费。消费者拉取消息消费者定期拉取Poll分配到的Partition消息处理后提交偏移量Offset。2. 消息流转图示MermaidConsumer (Group)Broker (Follower)Broker (Leader)ProducerConsumer (Group)Broker (Follower)Broker (Leader)Producer1. 生产者发送消息2. 消费者拉取消息发送消息到Topic-Partition0 (Key: order-1)写入本地日志 (LEO100)同步消息 (LEO100)确认同步 (LEO100)返回ACK (消息提交成功)Poll请求 (获取Partition0消息)返回消息 (Offset99, Value订单数据)处理消息 (扣减库存)提交偏移量 (Offset100)四、Kafka安装ZooKeeper传统模式CentOS 71. 环境准备CentOS 7系统关闭防火墙或开放端口2181、9092systemctl stop firewalld systemctl disable firewalld安装JDK 8yum install java-1.8.0-openjdk-devel -y2. 安装ZooKeeperKafka依赖步骤1下载并解压wget https://archive.apache.org/dist/zookeeper/zookeeper-3.7.1/apache-zookeeper-3.7.1-bin.tar.gztar -zxvf apache-zookeeper-3.7.1-bin.tar.gz -C /opt/mv /opt/apache-zookeeper-3.7.1-bin /opt/zookeeper步骤2配置ZooKeepercd /opt/zookeeper/confcp zoo_sample.cfg zoo.cfgvim zoo.cfg # 修改以下配置dataDir/var/lib/zookeeper # 数据存储目录clientPort2181 # 客户端端口步骤3启动ZooKeepermkdir -p /var/lib/zookeeper/opt/zookeeper/bin/zkServer.sh start # 启动/opt/zookeeper/bin/zkServer.sh status # 查看状态显示Mode: standalone为成功3. 安装Kafka Broker步骤1下载并解压wget https://archive.apache.org/dist/kafka/3.6.0/kafka_2.13-3.6.0.tgztar -zxvf kafka_2.13-3.6.0.tgz -C /opt/mv /opt/kafka_2.13-3.6.0 /opt/kafka步骤2配置Kafkacd /opt/kafka/configvim server.properties # 修改以下配置# 核心配置broker.id0 # 当前Broker唯一ID集群中不可重复listenersPLAINTEXT://localhost:9092 # 监听地址本地测试用localhostlog.dirs/var/lib/kafka/logs # 消息存储目录zookeeper.connectlocalhost:2181/kafka # 连接ZooKeeper/kafka为根节点步骤3启动Kafkamkdir -p /var/lib/kafka/logs/opt/kafka/bin/kafka-server-start.sh -daemon config/server.properties # 后台启动jps # 查看进程显示Kafka为成功4. 创建Topic测试用/opt/kafka/bin/kafka-topics.sh --create \--topic order-topic \ # 主题名称--bootstrap-server localhost:9092 \ # Kafka地址--partitions 3 \ # 分区数建议≥3--replication-factor 1 # 副本数单节点只能设1五、Spring Boot保姆级案例生产者消费者1. 项目结构src/main/java/com/example/kafkademo/├── KafkaDemoApplication.java # 启动类├── model/Order.java # 订单实体类├── producer/OrderProducer.java # 生产者服务├── consumer/OrderConsumer.java # 消费者服务└── controller/OrderController.java # 测试接口src/main/resources/└── application.yml # 配置文件2. pom.xml依赖?xml version1.0 encodingUTF-8?project xmlnshttp://maven.apache.org/POM/4.0.0 xmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsdmodelVersion4.0.0/modelVersionparentgroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-parent/artifactIdversion2.7.15/version !-- Spring Boot 2.7.x稳定版 --relativePath//parentgroupIdcom.example/groupIdartifactIdkafka-demo/artifactIdversion0.0.1-SNAPSHOT/versionnamekafka-demo/namedependencies!-- Web依赖提供HTTP接口 --dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-web/artifactId/dependency!-- Kafka依赖 --dependencygroupIdorg.springframework.kafka/groupIdartifactIdspring-kafka/artifactId/dependency!-- Lombok简化实体类 --dependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactIdoptionaltrue/optional/dependency/dependenciesbuildpluginsplugingroupIdorg.springframework.boot/groupIdartifactIdspring-boot-maven-plugin/artifactId/plugin/plugins/build/project3. application.yml配置server:port: 8080 # 应用端口spring:application:name: kafka-demo # 应用名称kafka:bootstrap-servers: localhost:9092 # Kafka集群地址多个用逗号分隔# 生产者配置producer:key-serializer: org.apache.kafka.common.serialization.StringSerializer # Key序列化器字符串value-serializer: org.springframework.kafka.support.serialization.JsonSerializer # Value序列化器JSONacks: all # 消息确认级别all所有ISR副本确认最高可靠性retries: 3 # 发送失败重试次数enable-idempotence: true # 启用幂等性防重复消息# 消费者配置consumer:group-id: order-group # 消费者组ID同一组内负载均衡key-deserializer: org.apache.kafka.common.serialization.StringDeserializer # Key反序列化器value-deserializer: org.springframework.kafka.support.serialization.JsonDeserializer # Value反序列化器auto-offset-reset: earliest # 无偏移量时策略earliest从头消费enable-auto-commit: false # 关闭自动提交偏移量手动控制properties:spring.json.trusted.packages: com.example.kafkademo.model # 信任的实体类包JSON反序列化用# 监听器配置消费者listener:ack-mode: manual_immediate # 手动立即提交偏移量处理完一条提交一条concurrency: 3 # 并发消费者数建议Topic分区数此处3分区4. 实体类Order.javapackage com.example.kafkademo.model;import lombok.AllArgsConstructor;import lombok.Data;import lombok.NoArgsConstructor;import java.math.BigDecimal;/*** 订单实体类消息载体*/Data // Lombok注解自动生成getter/setter/toString等NoArgsConstructor // 无参构造AllArgsConstructor // 全参构造public class Order {private String orderId; // 订单IDprivate String productName; // 商品名称private BigDecimal amount; // 订单金额private String status; // 订单状态CREATED/PAID/SHIPPED}5. 生产者服务OrderProducer.javapackage com.example.kafkademo.producer;import com.example.kafkademo.model.Order;import lombok.extern.slf4j.Slf4j;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.kafka.core.KafkaTemplate;import org.springframework.kafka.support.SendResult;import org.springframework.stereotype.Service;import org.springframework.util.concurrent.ListenableFuture;import org.springframework.util.concurrent.ListenableFutureCallback;import java.math.BigDecimal;import java.util.UUID;/*** 订单生产者服务发送订单消息到Kafka*/Service // 标记为Spring服务组件Slf4j // Lombok日志注解public class OrderProducer {// 注入KafkaTemplateSpring Boot自动配置用于发送消息Autowiredprivate KafkaTemplateString, Order kafkaTemplate;private static final String TOPIC_NAME order-topic; // 目标Topic名称需与消费者一致/*** 发送订单消息* param order 订单对象若为null则自动生成测试订单*/public void sendOrder(Order order) {// 1. 若订单ID为空生成UUID作为订单IDif (order null) {order new Order();order.setOrderId(UUID.randomUUID().toString()); // 随机生成订单ID