建筑企业网站模板,lnmt wordpress,购物网站html代码,网站美工SpringBoot集成Kafka实现消息发送和接收。 一、Kafka 简介二、Kafka 功能三、POM依赖四、配置文件五、生产者六、消费者 君子之学贵一#xff0c;一则明#xff0c;明则有功。 一、Kafka 简介
Kafka 是由 Apache 软件基金会开发的一个开源流处理平台#xff0c;最初由 Link… SpringBoot集成Kafka实现消息发送和接收。 一、Kafka 简介二、Kafka 功能三、POM依赖四、配置文件五、生产者六、消费者 君子之学贵一一则明明则有功。 一、Kafka 简介
Kafka 是由 Apache 软件基金会开发的一个开源流处理平台最初由 LinkedIn 公司开发并于 2011 年开源。它是一种高吞吐量的分布式发布 - 订阅消息系统以可持久化、高吞吐、低延迟、高容错等特性而著称。 Kafka 主要由生产者Producer、消费者Consumer、主题Topic、分区Partition和代理Broker等组件构成。生产者负责将数据发送到 Kafka 集群消费者从集群中读取数据。主题是一种逻辑上的分类数据被发送到特定的主题。每个主题又可以划分为多个分区以实现数据的并行处理和提高系统的可扩展性。代理则是 Kafka 集群中的服务器节点负责接收和存储生产者发送的数据并为消费者提供数据读取服务。
二、Kafka 功能
消息队列功能Kafka 可以作为消息队列使用在应用程序之间传递消息。生产者将消息发送到主题不同的消费者可以从主题中订阅并消费消息实现应用程序解耦。例如在电商系统中订单生成模块可以将订单消息发送到 Kafka 主题后续的库存管理、物流配送等模块可以从该主题消费订单消息各自独立处理降低模块间的耦合度。 数据存储功能Kafka 具有持久化存储能力它将消息数据存储在磁盘上并且通过多副本机制保证数据的可靠性。即使某个节点出现故障数据也不会丢失。这种特性使得 Kafka 不仅可以作为消息队列还能用于数据的长期存储和备份例如用于存储系统的操作日志方便后续的数据分析和故障排查。 流处理功能Kafka 可以与流处理框架如 Apache Flink、Spark Streaming 等集成对实时数据流进行处理。通过将实时数据发送到 Kafka 主题流处理框架可以从主题中读取数据并进行实时计算、分析和转换。例如在实时监控系统中通过 Kafka 收集服务器的性能指标数据然后使用流处理框架对这些数据进行实时分析及时发现性能异常并发出警报。
三、POM依赖 !-- kafka--dependencygroupIdorg.springframework.kafka/groupIdartifactIdspring-kafka/artifactIdversion2.8.11/version/dependency四、配置文件
spring:# Kafka 配置kafka:# Kafka 服务器地址和端口 代理地址可以多个bootstrap-servers: IP:9092# 生产者配置producer:# 发送失败时的重试次数retries: 3# 每次批量发送消息的数量调整为较小值batch-size: 1# 生产者缓冲区大小buffer-memory: 33554432# 消息 key 的序列化器将 key 序列化为字节数组key-serializer: org.apache.kafka.common.serialization.StringSerializer# 消息 value 的序列化器将消息体序列化为字节数组value-serializer: org.apache.kafka.common.serialization.StringSerializer# 消费者配置consumer:# 当没有初始偏移量或当前偏移量不存在时从最早的消息开始消费auto-offset-reset: earliest# 是否自动提交偏移量enable-auto-commit: true# 自动提交偏移量的时间间隔毫秒延长自动提交时间间隔auto-commit-interval: 1000# 消息 key 的反序列化器将字节数组反序列化为 keykey-deserializer: org.apache.kafka.common.serialization.StringDeserializer# 消息 value 的反序列化器将字节数组反序列化为消息体value-deserializer: org.apache.kafka.common.serialization.StringDeserializer五、生产者 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.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;/*** 生产者** author chenlei*/
Slf4j
Component
public class KafkaProducer {/*** KafkaTemplate*/Autowiredprivate KafkaTemplateString, String kafkaTemplate;/*** 发送消息到指定的 Kafka 主题并可指定分组信息** param topic 消息要发送到的 Kafka 主题* param message 要发送的消息内容*/public void sendMessage(String topic, String message) {// 使用 KafkaTemplate 发送消息将消息发送到指定的主题ListenableFutureSendResultString, String future kafkaTemplate.send(topic, message);future.addCallback(new ListenableFutureCallbackSendResultString, String() {Overridepublic void onSuccess(SendResultString, String result) {// 消息发送成功后的处理逻辑可根据需要添加log.info(已发送消息[ message ]其偏移量[ result.getRecordMetadata().offset() ]);}Overridepublic void onFailure(Throwable ex) {// 消息发送失败后的处理逻辑使用日志记录异常log.error(发送消息[ message ] 失败, ex);}});}
}六、消费者
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;/*** author 消费者* chenlei*/
Slf4j
Component
public class KafkaConsumer {/*** 监听 Kafka 主题方法。** param record 从 Kafka 接收到的 ConsumerRecord包含消息的键值对*/KafkaListener(topics {topic}, groupId consumer.group-id, concurrency 5)public void listen(ConsumerRecord?, ? record) {// 打印接收到的消息的详细信息log.info(接收到 Kafka 消息: 主题 {}, 分区 {}, 偏移量 {}, 键 {}, 值 {},record.topic(), record.partition(), record.offset(), record.key(), record.value());}
}
