个人备案转企业网站期间,wordpress 相关文章插件,wordpress机械模板下载地址,设计工作室名片一、Kafka 介绍
1. MQ 的作用
消息队列#xff08;Message Queue#xff0c;简称 MQ#xff09;是一种用于跨进程通信的技术#xff0c;核心功能是通过异步消息的方式实现系统之间的解耦。它在现代分布式系统中有着广泛的应用#xff0c;主要作用体现在以下三个方面Message Queue简称 MQ是一种用于跨进程通信的技术核心功能是通过异步消息的方式实现系统之间的解耦。它在现代分布式系统中有着广泛的应用主要作用体现在以下三个方面
异步处理
在传统的同步调用中生产者和消费者需要同时在线并且生产者在完成任务后才能继续执行其他工作。这种模式限制了系统的性能。而引入消息队列后生产者可以将任务提交到队列中消费者按需消费任务从而提升系统的吞吐量。
示例快递员送快递到客户家效率低下。而菜鸟驿站的出现让快递员只需将包裹放置在驿站客户可以根据自己的时间安排取件。这种方式大大提高了效率。
解耦
解耦是消息队列最重要的功能之一。服务之间通过消息队列传递数据而不是直接调用对方的服务接口这样可以有效降低系统的耦合度。
示例《Thinking in JAVA》原书是英文版但通过翻译社将内容翻译成多种语言满足不同读者的需求。翻译社起到了桥梁作用不同语言之间的沟通不再直接依赖于作者和读者。
削峰填谷
在高并发场景下系统往往会遇到流量高峰导致系统负载过重。通过消息队列可以将流量暂存并按固定速率处理从而避免系统崩溃。
示例长江每年都会涨水但通过三峡大坝的调节下游的出水速度保持稳定避免了洪水泛滥。
2. 为什么要用 Kafka
Kafka 是一种高吞吐量、低延迟、分布式的消息队列系统适合在大规模数据处理场景中使用。以下是 Kafka 的典型使用场景和优势
日志聚合场景
在大规模分布式系统中各个服务都会产生大量的日志信息。传统的日志收集方式往往存在以下问题
数据量大需要快速收集和处理来自各个渠道的海量日志。容错性要求高集群中允许少量节点出现故障而不影响整体服务。功能专注Kafka 专注于高吞吐量、低延迟的消息传递不追求复杂的消息处理功能。
核心优势
高吞吐量Kafka 能够处理数百万 TPS每秒事务处理量。低延迟通常在毫秒级别的延迟时间内完成消息传递。可扩展性通过增加节点和分区数量可以线性扩展处理能力。容错性通过副本机制保证消息的高可用性。持久化Kafka 使用磁盘存储消息保证消息的持久性。 二、Kafka 快速上手
1. 实验环境准备
要快速上手 Kafka首先需要搭建实验环境。以下是推荐的实验环境配置
虚拟机数量3 台操作系统CentOS 7Java 版本Java 8
环境配置步骤
下载 Kafka 和 Zookeeper。将 Kafka 解压到 /app/kafka 目录将 Zookeeper 解压到 /app/zookeeper 目录。配置环境变量确保系统能够识别 Kafka 和 Zookeeper 的命令。关闭防火墙以避免端口阻塞 systemctl stop firewalld.service2. 单机服务体验
为了更直观地理解 Kafka 的工作原理我们可以先体验单机版 Kafka 服务。
步骤 1启动 Zookeeper
Kafka 依赖 Zookeeper 进行元数据管理和选举机制。在实际部署中通常使用独立的 Zookeeper 集群。
启动 Zookeeper 服务
nohup bin/zookeeper-server-start.sh config/zookeeper.properties 检查 Zookeeper 是否正常启动
jps确认输出中有 QuorumPeerMain 进程。
步骤 2启动 Kafka
启动 Kafka 服务前需要确保 Zookeeper 服务正常运行。
启动 Kafka 服务
nohup bin/kafka-server-start.sh config/server.properties 确认 Kafka 是否正常启动
jps检查输出中是否包含 Kafka 进程。
步骤 3创建和使用 Topic
Kafka 的基础工作机制是通过 Topic 进行消息的传递。 创建 Topic bin/kafka-topics.sh --create --topic test --bootstrap-server localhost:9092发送消息 启动生产者端并发送消息 bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test这是一条测试消息消费消息 启动消费者端并接收消息 bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning3. 理解 Kafka 的消息传递机制
Kafka 的消息传递机制可以通过以下核心组件来理解
生产者Producer将消息发送到指定的 Topic。消费者Consumer从指定的 Topic 消费消息。Topic逻辑概念表示一类业务消息的集合。Partition物理概念实际存储消息的分区。BrokerKafka 服务器实例存储和管理 Partition。
Kafka 的设计目标是通过这些组件实现高效、可靠的消息传递满足企业级数据管道的需求。 四、Kafka 集群服务
1. 为什么要使用集群
单机部署的 Kafka 在性能上虽然已经非常出色但在实际生产环境中通常需要使用 Kafka 集群来进一步提升数据存储能力和系统的高可用性。集群可以解决以下问题
1.1 解决海量数据存储问题
单个 Broker 服务器的存储能力有限当数据量增长到一定程度时单机难以承载。通过集群部署可以将数据分散存储在多个 Broker 中从而提升整体存储能力。
1.2 提高系统容错能力
单机环境中如果 Broker 崩溃所有数据都会丢失。而集群环境下每个 Partition 都有多个副本即使部分 Broker 节点宕机系统依然可以正常运行保证数据的高可用性。 五、理解服务端的 Topic、Partition 和 Broker
Kafka 的核心架构由 Topic、Partition 和 Broker 组成这三者之间的关系至关重要
Topic一个逻辑的消息分类每个 Topic 包含多条消息。Partition每个 Topic 可以分成多个 Partition每个 Partition 是一个消息队列。BrokerKafka 的服务器实例负责存储 Partition 数据并处理客户端请求。
5.1 创建分布式 Topic 示例
bin/kafka-topics.sh --bootstrap-server worker1:9092 --create --replication-factor 2 --partitions 4 --topic distributedTopic5.2 查看 Topic 信息
bin/kafka-topics.sh --bootstrap-server worker1:9092 --describe --topic distributedTopic六、章节总结Kafka 集群的整体结构
通过前面的学习我们可以总结 Kafka 集群的整体结构
Topic 是逻辑概念Producer 和 Consumer 通过 Topic 进行消息传递。Partition 是实际存储单元保证数据分散存储和负载均衡。Broker 是 Kafka 的服务器实例存储 Partition 数据并处理客户端请求。Zookeeper 管理 Kafka 集群的元数据和选举过程。Controller 是 Kafka 集群的核心管理节点负责管理 Topic 和 Partition 的分配。 七、Spring Boot 实现 Kafka 消息有序性
为了保证 Kafka 的消息有序性可以使用 Spring Boot 和 Kafka 的整合来实现。在 Java 的 Spring Boot 项目中我们通过指定消息的 Key 和自定义分区器来确保消息发送到相同的 Partition从而实现有序性。
7.1 依赖配置
在 Maven 项目中引入 Kafka 的依赖
dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-kafka/artifactId
/dependency7.2 配置 KafkaProducer
创建 Kafka 的生产者配置类
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;import java.util.HashMap;
import java.util.Map;Configuration
public class KafkaProducerConfig {Beanpublic ProducerFactoryString, String producerFactory() {MapString, Object configProps new HashMap();configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, localhost:9092);configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);return new DefaultKafkaProducerFactory(configProps);}Beanpublic KafkaTemplateString, String kafkaTemplate() {return new KafkaTemplate(producerFactory());}
}7.3 发送有序消息
创建一个消息发送服务确保消息使用相同的 Key 发送到同一个 Partition
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;Service
public class KafkaProducerService {private static final String TOPIC test_topic;Autowiredprivate KafkaTemplateString, String kafkaTemplate;public void sendMessage(String key, String message) {kafkaTemplate.send(TOPIC, key, message);}
}7.4 自定义分区器可选
如果有更复杂的分区逻辑可以自定义分区器
import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;import java.util.Map;public class CustomPartitioner implements Partitioner {Overridepublic int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {// 自定义分区逻辑return Math.abs(key.hashCode()) % cluster.partitionCountForTopic(topic);}Overridepublic void close() {}Overridepublic void configure(MapString, ? configs) {}
}7.5 设置一个 Topic 对应一个 Partition 的方法
如果业务需求是保证某个 Topic 的消息全局有序可以在创建 Topic 时将 Partition 数量设置为 1从而保证所有消息存储在同一个 Partition 中实现全局有序。
创建一个 Partition 的 Topic
bin/kafka-topics.sh --create --topic singlePartitionTopic --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1在 Spring Boot 中发送消息到该 Topic
Service
public class KafkaSinglePartitionProducerService {private static final String TOPIC singlePartitionTopic;Autowiredprivate KafkaTemplateString, String kafkaTemplate;public void sendMessage(String message) {kafkaTemplate.send(TOPIC, message);}
}通过这种方式所有发送到 singlePartitionTopic 的消息都会进入同一个 Partition确保消息顺序性。
