Kafka两种配置文件方式

1.yml配置文件（简单配置）

spring:kafka:bootstrap-servers: ip:端口consumer:group-id: group-testenable-auto-commit: trueauto-commit-interval: 1000msauto-offset-reset: latestkey-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.apache.kafka.common.serialization.StringDeserializer

2.类配置方式（使用了SASL/PLAINTEXT安全认证协议）

出现[Consumer clientId=consumer-1, groupId=group1] Bootstrap broker xxx (id: -1 rack: null) disconnected报错信息就需要用这种配置方式，加入安全认证

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;import java.util.HashMap;
import java.util.Map;/*** @author 向振华* @date 2021/05/10 15:58*/
@Configuration
@EnableKafka
public class KafkaConsumerConfig {@Beanpublic KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();factory.setConsumerFactory(consumerFactory());factory.setConcurrency(10);factory.getContainerProperties().setPollTimeout(1500);return factory;}public ConsumerFactory<String, String> consumerFactory() {return new DefaultKafkaConsumerFactory<>(consumerConfigs());}public Map<String, Object> consumerConfigs() {Map<String, Object> propsMap = new HashMap<>(16);// 服务地址propsMap.put("bootstrap.servers", "ip:端口");// 安全认证协议propsMap.put("security.protocol", "SASL_PLAINTEXT");propsMap.put("sasl.mechanism", "PLAIN");// 填充安全认证用户名和密码propsMap.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"usnm\" password=\"pwd\";");propsMap.put("group.id", "group-test");propsMap.put("enable.auto.commit", true);propsMap.put("auto.commit.interval.ms", 1000);// latest: 从最新的偏移量开始消费propsMap.put("auto.offset.reset", "latest");// 反序列化方式propsMap.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");propsMap.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");return propsMap;}
}

再加一个配置文件sasl.jaas.config到resource

KafkaServer {org.apache.kafka.common.security.plain.PlainLoginModule requiredusername="kafkaadmin"password="kafkaadminpwd"user_kafkaadmin="kafkaadminpwd"user_kafkaclient1="kafkaclient1pwd"user_kafkaclient2="kafkaclient2pwd";
};

使用：

@Slf4j
@Component
public class KafkaConsumer {@KafkaListener(topics = {"xxx"})public void listener(ConsumerRecord<?, ?> record) {log.info("收到消息 ---> " + record.value().toString());}
}

Kafka两种配置文件方式相关推荐

Spark对Kafka两种连接方式的对比——Receiver和Direct
在知乎 Flink 取代 Spark Streaming 的实战之路中,提到因此下面对两种方式进行详细说明一下. Receiver方式 Receiver:接收器模式是使用Kafka高级Consume ...
Kafka结合Spark-streaming 的两种连接方式(AWL与直连)
kafka结合spark-streaming的用法及说明之前博客有些,这里就不赘述了. 这篇文章说下他们结合使用的两种连接方式.(AWL与直连) 先看一张图: 这是kafka与streaming结合的 ...
SpringBoot的properties和yml两种配置方式, 配置注入参数, 以及配置文件读取失效的问题
SpringBoot支持两种配置方式,一种是properties文件,一种是yml 首先在pom文件中添加依赖: <dependency><groupId>org.spring ...
redis的两种持久化方式详解
一.背景在实际开发中,为了保证数据的完整性,防止数据丢失,我们除了在原有的传统数据库保存数据的同时,最好是再用redis持久化再保存一次数据.如果仅仅是使用redis而不进行持久化配置的话,当red ...
android旋转动画的两种实现方式
在android开发,我们会常常使用到旋转动画,普通情况下旋转动画有两种实现方式,一种是直接通过java代码去实现,第二种是通过配置文件实现动画.以下是两种动画的基本是用法: 纯Java代码实现: / ...
wdcp支持两种安装方式
v3.2版本已发布,支持多PHP版本共存共用,支持SSL证书,更多可看论坛 v3版讨论区更多安装说明请看 http://www.wdlinux.cn/bbs/thread-57643-1-1.htm ...
Java框架篇---spring aop两种配置方式
Java框架篇---spring aop两种配置方式第一种:注解配置AOP 注解配置AOP(使用 AspectJ 类库实现的),大致分为三步: 1. 使用注解@Aspect来定义一个切面,在切面中 ...
探究Redis两种持久化方式下的数据恢复
对长期奋战在一线的后端开发人员来说,都知道redis有两种持久化方式RDB和AOF,虽说大家都知道这两种方式大概运作方式,但想必有实操的人不会太多. 这里是自己实操两种持久化方式的一点点记录. 先看以 ...
Destoon数据库配置文件在哪_Mybatis 系列 2：Mybatis 的两种配置文件
Mybatis 的配置文件,有两种: 一.Mybatis 全局配置文件(主配置文件): 起名:不固定,但一般起名要见名知意 --> mybatis-config.xml 路径:classpath ...

Kafka两种配置文件方式

Kafka两种配置文件方式相关推荐

最新文章

热门文章