kafka springBoot配置的實現(xiàn)
1、properties 配置
control.command.kafka.enabled=true control.command.kafka.bootstrap-servers=172.0.0.1:9092 control.command.kafka.command-topics=lastTopic control.command.kafka.consumer.group-id=consumer-eslink-iwater-control-command control.command.kafka.consumer.properties.session.timeout.ms=30000 control.command.kafka.consumer.properties.request.timeout.ms=90000 control.command.kafka.consumer.fetch-min-size=10KB control.command.kafka.consumer.fetch-max-wait=500 control.command.kafka.consumer.max-poll-records=1000 control.command.kafka.consumer.auto-offset-reset=earliest control.command.kafka.listener.ack-mode=MANUAL_IMMEDIATE control.command.kafka.listener.concurrency=1 control.command.kafka.listener.type=SINGLE control.command.kafka.producer.acks=all control.command.kafka.producer.batchSize=4096 control.command.kafka.producer.bufferMemory=40960 control.command.kafka.producer.linger=10 control.command.kafka.producer.retries=3
2、Config 配置
@Configuration
@ConditionalOnExpression("${control.command.kafka.enabled:false}")
@EnableKafka
public class ControlCommandKafkaConfig {
@Bean("controlCommandKafkaProperties")
@ConfigurationProperties("control.command.kafka")
@Primary
public KafkaProperties kafkaProperties() {
return new KafkaProperties();
}
@Bean("controlCommandKafkaConsumerFactory")
public ConsumerFactory<Object, Object> kafkaConsumerFactory(
@Qualifier("controlCommandKafkaProperties") KafkaProperties inProps) {
DefaultKafkaConsumerFactory<Object, Object> consumerFactory = new DefaultKafkaConsumerFactory<>(inProps.buildConsumerProperties());
return consumerFactory;
}
@Bean("controlCommandBatchFactory")
@DependsOn("controlCommandKafkaProperties")
public KafkaListenerContainerFactory<?> egBatchFactory(@Qualifier("controlCommandKafkaProperties") KafkaProperties inProps,
@Qualifier("controlCommandKafkaConsumerFactory") ConsumerFactory consumerFactory) {
ConcurrentKafkaListenerContainerFactory<Object, Object> listenerFactory =
new ConcurrentKafkaListenerContainerFactory<>();
listenerFactory.setConsumerFactory(consumerFactory);
configureListenerFactory(listenerFactory, inProps);
configureContainer(listenerFactory.getContainerProperties(), inProps);
return listenerFactory;
}
@Bean("controlCommandKafkaTemplate")
public KafkaTemplate<String, String> kafkaTemplate(@Qualifier("controlCommandKafkaProperties") KafkaProperties inProps) {
return new KafkaTemplate(new DefaultKafkaProducerFactory(inProps.buildProducerProperties()));
}
private void configureListenerFactory(ConcurrentKafkaListenerContainerFactory<Object, Object> factory, KafkaProperties inProps) {
PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
KafkaProperties.Listener properties = inProps.getListener();
// 設(shè)置監(jiān)聽線程并發(fā)數(shù) control.command.kafka.listener.concurrency
map.from(properties::getConcurrency).to(factory::setConcurrency);
// control.kafka.listener.type=batch 時 批量監(jiān)聽
if (properties.getType().equals(KafkaProperties.Listener.Type.BATCH)) {
factory.setBatchListener(true);
}
}
private void configureContainer(ContainerProperties container, KafkaProperties inProps) {
PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
KafkaProperties.Listener properties = inProps.getListener();
map.from(properties::getAckMode).to(container::setAckMode);
map.from(properties::getClientId).to(container::setClientId);
map.from(properties::getAckCount).to(container::setAckCount);
map.from(properties::getAckTime).as(Duration::toMillis).to(container::setAckTime);
map.from(properties::getPollTimeout).as(Duration::toMillis).to(container::setPollTimeout);
map.from(properties::getNoPollThreshold).to(container::setNoPollThreshold);
map.from(properties::getIdleEventInterval).as(Duration::toMillis).to(container::setIdleEventInterval);
map.from(properties::getMonitorInterval).as(Duration::getSeconds).as(Number::intValue)
.to(container::setMonitorInterval);
map.from(properties::getLogContainerConfig).to(container::setLogContainerConfig);
map.from(properties::isMissingTopicsFatal).to(container::setMissingTopicsFatal);
}
}
2.1 @EnableKafka 注解
‘@EnableKafka’ 是用于在 Spring Boot 應(yīng)用程序中啟用 Apache Kafka 的注解。當(dāng)你在 Spring Boot 應(yīng)用程序的配置類上添加 @EnableKafka 注解時，它會激活 Kafka 基礎(chǔ)設(shè)施，使你能夠在應(yīng)用程序中使用 Kafka 相關(guān)的組件。
以下是對這個注解的簡要解釋：
基礎(chǔ)設(shè)施激活：@EnableKafka 注解告訴 Spring 在應(yīng)用程序中設(shè)置所需的 Kafka 基礎(chǔ)設(shè)施。這包括創(chuàng)建必要的 Kafka bean 和配置。
Kafka Template：在啟用 Kafka 后，你可以使用 Spring Kafka 提供的 KafkaTemplate 類輕松地向 Kafka 主題發(fā)送消息。KafkaTemplate 抽象了底層的 Kafka 生產(chǎn)者 API，簡化了消息發(fā)送的過程。
消息監(jiān)聽器：通過 @EnableKafka，你還可以使用 Spring 的 @KafkaListener 注解設(shè)置 Kafka 消息消費(fèi)者。@KafkaListener 注解應(yīng)用于 Spring 組件中的方法，使該方法可以作為 Kafka 消息監(jiān)聽器。當(dāng) Kafka 主題中有可用的消息時，帶有 @KafkaListener 注解的方法將自動被調(diào)用，并處理消息內(nèi)容。
下面是在 Spring Boot 應(yīng)用程序中使用 @EnableKafka 的示例代碼：
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.kafka.annotation.EnableKafka;
@SpringBootApplication
@EnableKafka
public class MyKafkaApplication {
public static void main(String[] args) {
SpringApplication.run(MyKafkaApplication.class, args);
}
}
通過這樣的配置，你的 Spring Boot 應(yīng)用程序?qū)⒂?Kafka 支持，你可以使用 KafkaTemplate 進(jìn)行消息發(fā)送，使用 @KafkaListener 進(jìn)行消息消費(fèi)。請確保在 application.properties 或 application.yml 文件中配置 Kafka 相關(guān)屬性，例如 Kafka 代理地址和其他必要的配置。
請注意，為了使 @EnableKafka 正常工作，你需要在項目的構(gòu)建配置中包含所需的 Kafka 依賴項。對于 Maven，你可以添加以下依賴項：
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
一旦 Kafka 正在運(yùn)行，并且應(yīng)用程序正確配置了 @EnableKafka，你就可以在 Spring Boot 應(yīng)用程序中構(gòu)建基于 Kafka 的消息傳遞功能。
2.2、@ConditionalOnExpression 注解
@ConditionalOnExpression 是 Spring Boot 中的一個條件注解之一。它允許你根據(jù)給定的 SpEL 表達(dá)式來決定是否啟用或禁用某個 Bean 或配置。
條件注解可以用于在 Spring Boot 應(yīng)用程序中根據(jù)特定條件來動態(tài)創(chuàng)建 Bean 或配置，從而根據(jù)不同的配置或環(huán)境來靈活地管理應(yīng)用程序的行為。
@ConditionalOnExpression 的工作方式是：它在配置類或 Bean 上進(jìn)行標(biāo)記，然后在應(yīng)用程序啟動過程中解析 SpEL 表達(dá)式。如果 SpEL 表達(dá)式的結(jié)果為 true，則相關(guān)的 Bean 或配置將被啟用，否則將被禁用。
以下是 @ConditionalOnExpression 的示例使用方式：
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Conditional;
import org.springframework.context.annotation.Configuration;
@Configuration
public class MyConfiguration {
@Bean
@ConditionalOnExpression("${myapp.feature.enabled:true}")
public MyBean myBean() {
// 返回需要創(chuàng)建的 Bean 實例
return new MyBean();
}
}
2.3、 @RefreshScope 注解
@RefreshScope 是 Spring Cloud 中的一個注解，用于實現(xiàn)動態(tài)刷新 Spring Bean 的配置信息。它通常與 Spring Cloud Config 配合使用，能夠在運(yùn)行時更新應(yīng)用程序配置，而無需重啟應(yīng)用。
在微服務(wù)架構(gòu)中，使用 Spring Cloud Config 可以將配置信息集中管理，然后通過 @RefreshScope 注解實現(xiàn)配置的動態(tài)刷新。這使得應(yīng)用程序在運(yùn)行時可以獲取最新的配置信息，而不需要停止和啟動應(yīng)用。
2.4、@DependsOn 注解
@DependsOn 是 Spring Framework 中的一個注解，用于指定 Spring Bean 之間的依賴關(guān)系。通過在 Bean 上添加 @DependsOn 注解，你可以確保指定的 Bean 會在其所依賴的其他 Bean 初始化之后再進(jìn)行初始化。
當(dāng)一個 Bean 希望在另一個 Bean 初始化完成后再初始化時，可以使用 @DependsOn 注解來定義這種依賴關(guān)系。這對于確保 Bean 之間的正確順序初始化非常有用，特別是當(dāng)某些 Bean 需要依賴其他 Bean 才能正確地進(jìn)行初始化或工作時。
2.5、 listener.ack-mode 消息的確認(rèn)模式
在 Spring Kafka 中，AckMode 是用于配置消息消費(fèi)者的消息確認(rèn)模式（Acknowledgment Mode）。這個枚舉類型用于決定在消費(fèi)者處理完 Kafka 消息后如何向 Kafka 服務(wù)器發(fā)送確認(rèn)，告知服務(wù)器消息是否已經(jīng)被成功消費(fèi)。
Spring Kafka 支持以下幾種 AckMode：
AUTO: 這是默認(rèn)的確認(rèn)模式。在這種模式下，消費(fèi)者會自動在處理完消息后向 Kafka 服務(wù)器發(fā)送確認(rèn)。當(dāng)消費(fèi)者成功處理消息后，Kafka 服務(wù)器會將偏移量（offset）移動到下一條消息，表示該消息已成功消費(fèi)。但是在這種模式下，如果處理消息時發(fā)生異常，Kafka 服務(wù)器會重新發(fā)送相同的消息，可能會導(dǎo)致消息的重復(fù)消費(fèi)。
MANUAL: 在這種模式下，消費(fèi)者需要手動管理消息的確認(rèn)。當(dāng)消費(fèi)者成功處理消息后，需要調(diào)用 Acknowledgment 對象的 acknowledge() 方法，手動向 Kafka 服務(wù)器發(fā)送確認(rèn)。這樣可以確保消息的準(zhǔn)確處理，避免重復(fù)消費(fèi)。
MANUAL_IMMEDIATE: 這是另一種手動確認(rèn)模式，與 MANUAL 模式類似。區(qū)別在于，當(dāng)消費(fèi)者調(diào)用 acknowledge() 方法時，它會立即提交確認(rèn)而不是等待下一次輪詢。這樣可以更快地將確認(rèn)提交給 Kafka 服務(wù)器。
你可以通過在 @KafkaListener 注解中設(shè)置 AckMode 來配置消息消費(fèi)者的確認(rèn)模式。例如：
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
@KafkaListener(topics = "my_topic", groupId = "my_group", ackMode = "MANUAL")
public void processMessage(String message, Acknowledgment acknowledgment) {
// 處理消息的邏輯
// 手動確認(rèn)消息
acknowledgment.acknowledge();
}
在上面的例子中，我們將 ackMode 設(shè)置為 MANUAL，表明消息的確認(rèn)模式為手動確認(rèn)。在處理消息后，我們手動調(diào)用 acknowledgment.acknowledge() 來確認(rèn)消息的消費(fèi)。
選擇合適的 AckMode 取決于你的應(yīng)用程序需求和消費(fèi)者的可靠性要求。如果應(yīng)用程序?qū)ο⒌闹貜?fù)消費(fèi)有一定的容忍度，并且希望簡化消費(fèi)者的處理邏輯，可以選擇 AUTO 模式。如果應(yīng)用程序?qū)ο⒌臏?zhǔn)確性要求較高，并且愿意手動確認(rèn)消息的處理，可以選擇 MANUAL 或 MANUAL_IMMEDIATE 模式。
2.6、kafka.consumer.auto-offset-reset 偏移量
kafka.consumer.auto-offset-reset 是 Kafka 消費(fèi)者的一個重要配置屬性，它決定了當(dāng)一個新的消費(fèi)者加入消費(fèi)者組或者消費(fèi)者在某個分區(qū)上沒有有效的偏移量時，消費(fèi)者應(yīng)該從何處開始消費(fèi)消息。
這個屬性有以下幾個可能的值：
rliest: 如果消費(fèi)者在某個分區(qū)上沒有有效的偏移量，或者消費(fèi)者組第一次加入時，從最早的可用偏移量開始消費(fèi)消息。換句話說，從分區(qū)的起始位置開始消費(fèi)。
latest: 如果消費(fèi)者在某個分區(qū)上沒有有效的偏移量，或者消費(fèi)者組第一次加入時，從最新的可用偏移量開始消費(fèi)消息。換句話說，只消費(fèi)自加入后產(chǎn)生的新消息。
none: 如果消費(fèi)者在某個分區(qū)上沒有有效的偏移量，拋出一個異常。這種情況下，消費(fèi)者必須手動設(shè)置初始偏移量。
anything else: 拋出一個異常。
這個屬性通常在 Kafka 消費(fèi)者的配置中使用，用來控制消費(fèi)者在特定情況下的起始消費(fèi)位置。你可以根據(jù)你的業(yè)務(wù)需求來選擇適合的值。如果你希望消費(fèi)者能夠從最早的消息開始消費(fèi)，以確保不錯過任何消息，可以將它設(shè)置為 earliest。如果你只關(guān)心新產(chǎn)生的消息，可以將它設(shè)置為 latest。
3、KafkaProducer 生產(chǎn)者
@Component
@RefreshScope
@ConditionalOnExpression("${control.command.kafka.enabled:false}")
@DependsOn(value = {"controlCommandKafkaTemplate"})
public class ControlCommandKafkaProducer extends BaseLogable {
@Resource(name = "controlCommandKafkaTemplate")
private KafkaTemplate<String, String> controlCommandKafkaTemplate;
public void send(String key, String topic, String msg) {
bizLogger.info("sent kafka msg, topic:{}, key: {}, msg: {}", topic, key, msg);
ListenableFuture<SendResult<String, String>> listenableFuture = controlCommandKafkaTemplate.send(topic, key, msg);
listenableFuture.addCallback(result -> bizLogger.info("sent msg success：{}", msg),
e -> {
bizLogger.error("sent msg failure, msg: {}", msg, e);
});
}
}
3.1 org.springframework.kafka.core.KafkaTemplate 使用
3.1.1 發(fā)送消息
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
@Service
public class KafkaProducerService {
private final KafkaTemplate<String, String> kafkaTemplate;
public KafkaProducerService(KafkaTemplate<String, String> kafkaTemplate) {
this.kafkaTemplate = kafkaTemplate;
}
public void sendMessage(String topic, String message) {
kafkaTemplate.send(topic, message);
}
}
3.1.2 發(fā)送消息并指定分區(qū)
public void sendMessageToPartition(String topic, int partition, String message) {
kafkaTemplate.send(topic, partition, null, message);
}
3.1.3 發(fā)送消息并指定鍵
public void sendMessageWithKey(String topic, String key, String message) {
kafkaTemplate.send(topic, key, message);
}
3.1.4 發(fā)送消息并等待確認(rèn)
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.kafka.support.SendResult;
import org.springframework.kafka.support.future.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
public void sendMessageAndWaitForConfirmation(String topic, String message) {
ProducerRecord<String, String> record = new ProducerRecord<>(topic, message);
ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(record);
future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
@Override
public void onSuccess(SendResult<String, String> result) {
// 消息發(fā)送成功的處理邏輯
}
@Override
public void onFailure(Throwable ex) {
// 消息發(fā)送失敗的處理邏輯
}
});
}
4、KafkaConsumer 消費(fèi)者
4.1 單條消費(fèi)
單條消費(fèi)時配置
control.command.kafka.listener.type=SINGLE
@Component
@RefreshScope
@ConditionalOnExpression("${control.command.kafka.enabled:false}")
@DependsOn(value = {"controlCommandBatchFactory"})
public class ControlCommandKafkaConsumer extends BaseLogable {
/**
* 監(jiān)聽主題，單條消費(fèi)
*/
@KafkaListener(id = "${control.command.kafka.consumer.groupId}", topics = "${kafka.command.topic.ecgs-command-up}", containerFactory = "controlCommandBatchFactory")
public void listen(ConsumerRecord<String, String> record, Acknowledgment ack) {
bizLogger.info("kafka msg::: " + record.value() + ", key: " + record.key());
//接收Kafka消息
String data = record.value();
//接收消息確認(rèn)
ack.acknowledge();
}
}
4.2 多條消費(fèi)
多條消費(fèi)時配置
control.command.kafka.listener.type=BATCH
package cc.eslink.yq.iwater.kafka;
import cc.eslink.common.base.BaseLogable;
import cc.eslink.yq.iwater.dto.schedule.AlarmInfoDTO;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.context.annotation.DependsOn;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import java.util.List;
@Component
@RefreshScope
@ConditionalOnExpression("${control.command.kafka.enabled:false}")
@DependsOn(value = {"controlCommandBatchFactory"})
public class ControlCommandKafkaConsumer extends BaseLogable {
@KafkaListener(id = "${control.command.kafka.consumer.groupId}", topics = "${kafka.command.topic.ecgs-command-up}", containerFactory = "controlCommandBatchFactory")
public void listen(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
bizLogger.info("== 接收到 kafka msg 條數(shù):::{} ==", records.size());
//接收消息確認(rèn)
ack.acknowledge();
//接收Kafka消息
for (ConsumerRecord<String, String> record : records) {
try {
final AlarmInfoDTO alarmInfoDTO = JSON.parseObject(record.value(), new TypeReference<AlarmInfoDTO>() {});
} catch (Exception e) {
expLogger.error("alarm.dispose error", record, e);
}
}
}
}到此這篇關(guān)于kafka springBoot配置的實現(xiàn)的文章就介紹到這了,更多相關(guān)kafka springBoot 配置內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
使用Maven打包的時候，可能會因為單元測試打包失敗，這時候就需要跳過單元測試。本文就介紹了Maven打包跳過測試的實現(xiàn)方法，感興趣的可以了解一下2021-06-06
Maven dependencyManagement元素標(biāo)簽的具體使用
在Maven中dependencyManagement的作用其實相當(dāng)于一個對所依賴jar包進(jìn)行版本管理的管理器,本文主要介紹了Maven dependencyManagement元素標(biāo)簽的具體使用,感興趣的可以了解一下2024-03-03
這篇文章主要介紹了Java的抽象方法和抽象類,結(jié)合實例形式詳細(xì)分析了java抽象方法和抽象類的相關(guān)原理、使用方法及操作注意事項,需要的朋友可以參考下2019-09-09
Java創(chuàng)建對象的六種常用方式總結(jié)
作為Java開發(fā)者,經(jīng)常創(chuàng)建很多對象,你是否知道Java中創(chuàng)建對象有哪些方式呢?下面這篇文章主要給大家介紹了關(guān)于Java創(chuàng)建對象的六種常用方式,文中通過代碼介紹的非常詳細(xì),需要的朋友可以參考下2024-06-06
這篇文章主要介紹了IntelliJ?IDEA如何集成maven問題,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2023-12-12
顧名思義,垃圾收集(Garbage?Collection)的意思就是?——?找到垃圾并進(jìn)行清理。但現(xiàn)有的垃圾收集實現(xiàn)卻恰恰相反:?垃圾收集器跟蹤所有正在使用的對象,并把其余部分當(dāng)做垃圾。記住這一點以后,?我們再深入講解內(nèi)存自動回收的原理，探究?JVM?中垃圾收集的具體實現(xiàn)2022-01-01
Java調(diào)用基于Ollama本地大模型的實現(xiàn)
本文主要介紹了Java調(diào)用基于Ollama本地大模型的實現(xiàn),實現(xiàn)文本生成、問答、文本分類等功能,開發(fā)者可以輕松配置和調(diào)用模型,具有一定的參考價值,感興趣的可以了解一下2025-03-03
解讀System.getProperty("ENM_HOME")中的值從哪獲取的
這篇文章主要介紹了解讀System.getProperty("ENM_HOME")中的值從哪獲取的問題,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2023-12-12
aven提供了一套強(qiáng)大的版本管理機(jī)制,允許開發(fā)者管理項目的版本號,并在不同的版本之間進(jìn)行升級和降級,以下是如何使用Maven進(jìn)行版本管理的詳細(xì)步驟和代碼示例,感興趣的小伙伴跟著小編一起來看看吧2024-11-11
解析web.xml中在Servlet中獲取context-param和init-param內(nèi)的參數(shù)
本篇文章是對web.xml中在Servlet中獲取context-param和init-param內(nèi)的參數(shù)進(jìn)行了詳細(xì)的分析介紹，需要的朋友參考下2013-07-07