SpringBoot项目集成kafka及配置连接账号密码

susu1083018911

2058人浏览 · 2024-06-06 09:59:39

susu1083018911 · 2024-06-06 09:59:39 发布

kafka的鉴权协议

PLAINTEXT

这是 Kafka 默认的协议,不提供任何认证机制。所有客户端都可以不经认证就连接到 Kafka 集群。
在生产环境中,强烈不建议使用这种协议,因为它不提供任何安全保障。

SSL

SSL 提供了一种加密通信的方式,可以保护客户端和服务器之间的数据传输。
SSL 需要在 Kafka 代理(Broker)和客户端之间建立信任关系,通常是通过 SSL 证书来实现的。
SSL 可以单独使用,也可以与其他认证协议(如 SASL)结合使用,提供更高级别的安全性。

SASL/PLAIN

SASL/PLAIN 是一种基于用户名和密码的简单认证机制。
客户端需要提供有效的用户名和密码才能连接到 Kafka 集群。
SASL/PLAIN 通常与 SSL 协议结合使用,以防止明文传输用户名和密码。

SASL/SCRAM

SASL/SCRAM 是一种更安全的基于挑战-响应的认证机制。
SASL/SCRAM 支持两种变体:SCRAM-SHA-256 和 SCRAM-SHA-512。
SCRAM 使用 HMAC 和随机数来对密码进行哈希计算,而不会在网络上传输明文密码。
SCRAM 通常也与 SSL 结合使用,以提供更高的安全性。

SASL/GSSAPI (Kerberos)

SASL/GSSAPI 是基于 Kerberos 的认证协议。
Kerberos 是一种广泛使用的网络认证协议,提供了集中式的认证和授权服务。
Kerberos 使用票据(Ticket)系统来验证客户端身份,而不是直接传输密码。
Kerberos 认证通常被认为是企业级环境中最安全的认证方式之一。

SASL/OAUTHBEARER

SASL/OAUTHBEARER 是一种使用 OAuth 2.0 访问令牌(Access Token)进行认证的协议。
OAuth 2.0 是一种广泛使用的授权框架,常用于授权第三方应用访问用户数据。
SASL/OAUTHBEARER 需要客户端获取有效的 OAuth 2.0 访问令牌,并在连接 Kafka 时提供该令牌进行认证。

引入依赖，不用配置version

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

KafkaConfiguration 公共配置

@Data
@Configuration
public class KafkaConfiguration {
 
    /**
     * 主机地址
     */
    @Value("${kafka.server-host}")
    private String bootstrapServers;
 
    /**
     * sasl 认证账号
     */
    @Value("${iot.kafka.sasl.username:admin}")
    private String userName;
 
    /**
     * sasl 密码
     */
    @Value("${iot.kafka.sasl.password:iot@2021}")
    private String password;
 
 
}

KafkaConsumerConfiguration 消费者配置

@Data
@Configuration
public class KafkaConsumerConfiguration {
 
    /**
     * 默认组id
     */
    @Value("${iot.kafka.consumer.properties.group-id:default-group}")
    private String groupId;
 
    @Value("${iot.kafka.consumer.properties.fetch-max-wait:5000}")
    private Integer fetchMaxWait;
 
    /**
     * 此设置限制每次调用poll返回的消息数，这样可以更容易的预测每次poll间隔要处理的最大值。通过调整此值，可以减少poll间隔，减少重新平衡分组的
     */
    @Value("${iot.kafka.consumer.properties.max-poll-record:100}")
    private Integer maxPollRecordsConfig;
 
    /**
     * 增大poll的间隔，可以为消费者提供更多的时间去处理返回的消息（调用poll（long）返回的消息，通常返回的消息都是一批），缺点是此值越大将会延迟组重新平衡。
     */
    @Value("${iot.kafka.consumer.properties.max-poll-interval-ms:100}")
    private Integer maxPollIntervalConfig;
 
    /**
     * 是否开启自动提交
     */
    @Value("${iot.kafka.consumer.properties.enable-auto-commit:#{false}}")
    private boolean enableAutoCommitConfig;
 
    /**
     * 消费策略
     * earliest  当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
     * latest 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
     * none topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
     */
    @Value("${iot.kafka.consumer.properties.auto-offset-reset:earliest}")
    private String autoOffsetResetConfig;
 
    @Value("${iot.kafka.consumer.properties.auto-commit-interval:1000}")
    private String autoCommitIntervalMsConfig;
 
    @Value("${iot.kafka.consumer.properties.session-timeout:30000}")
    private String sessionTimeoutMsConfig;
 
}

KafkaListenerConfiguration 监听配置

@Data
@Configuration
public class KafkaListenerConfiguration {
 
    /**
     * 启用线程数（提高并发）
     */
    @Value("${iot.kafka.listener.concurrency:3}")
    private Integer concurrency;
 
    /**
     * 手动提交的方式，当enable-auto-commit: false时起作用
     * manual:手动调用Acknowledgment.acknowledge()后立即提交
     * record:当每一条记录被消费者监听器（ListenerConsumer）处理之后提交
     * batch:当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后提交
     * time: 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，距离上次提交时间大于TIME时提交
     * count:当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，被处理record数量大于等于COUNT时提交
     * count_time:当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后, 手动调用Acknowledgment.acknowledge()后提交
     */
    @Value("${iot.kafka.listener.ack-mode:manual_immediate}")
    private String ackMode;
 
    /**
     * 消费超时时间
     */
    @Value("${iot.kafka.listener.poll-timeout:3000}")
    private Long pollTimeout;
 
    /**
     * 是否开启批量处理
     */
    @Value("${iot.kafka.listener.batch_listener:#{true}}")
    private Boolean batchListener;
 
}

KafkaProducerConfiguration 生产者配置

@Data
@Configuration
public class KafkaProducerConfiguration {
    /**
     * 重试次数 默认值0
     */
    @Value("${iot.kafka.producer.retries:0}")
    private Integer retries;
 
    /**
     * acks = 0 如果设置为零，则生产者将不会等待来自服务器的任何确认，该记录将立即添加到套接字缓冲区并视为已发送。在这种情况下，无法保证服务器已收到记录，并且重试配置将不会生效（因为客户端通常不会知道任何故障），为每条记录返回的偏移量始终设置为 - 1。
     * acks = 1 这意味着leader会将记录写入其本地日志，但无需等待所有副本服务器的完全确认即可做出回应，在这种情况下，如果leader在确认记录后立即失败，但在将数据复制到所有的副本服务器之前，则记录将会丢失。
     * acks = all 这意味着leader将等待完整的同步副本集以确认记录，这保证了只要至少一个同步副本服务器仍然存活，记录就不会丢失，这是最强有力的保证，这相当于acks = -1 的设置。
     */
    @Value("${iot.kafka.producer.acks:all}")
    private String acks;
 
    /**
     * 指定缓存的大小，生产者缓存每个分区未发送的消息。默认 16384
     */
    @Value("${iot.kafka.producer.batch-size:16384}")
    private Integer batchSize;
 
    /**
     * 生产者发送请求之前等待一段时间，设置等待时间是希望更多地消息填补到未满的批中。 默认 30
     */
    @Value("${iot.kafka.producer.properties.linger.ms:30}")
    private Integer lingerMs;
 
    /**
     * 通过KafkaProducer发送出去的消息都是先进入到客户端本地的内存缓冲里，然后把很多消息收集成一个一个的Batch，再发送到Broker上去的 默认32m
     */
    @Value("${iot.kafka.producer.buffer-memory:33554432}")
    private Integer bufferMemory;
 
}

工厂配置

ConsumerFactoryBuilder 消费者工厂

@Configuration
public class ConsumerFactoryBuilder {
 
    @Autowired
    private KafkaConfiguration kafkaConfiguration;
 
    @Autowired
    private KafkaConsumerConfiguration kafkaConsumerConfiguration;
 
    @Autowired
    private KafkaListenerConfiguration kafkaListenerConfiguration;
 
    /**
     * 消费者配置
     *
     * @return properties
     */
    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new ConcurrentHashMap<>();
        //配置地址
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConfiguration.getBootstrapServers());
        //消费者组 默认组id
        props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaConsumerConfiguration.getGroupId());
        //是否开启自动提交
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, kafkaConsumerConfiguration.isEnableAutoCommitConfig());
        /* 消费策略
         * earliest  当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
         * latest 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
         * none topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
         */
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, kafkaConsumerConfiguration.getAutoOffsetResetConfig());
        //消费者默认等待服务响应时间(毫秒)
        props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, kafkaConsumerConfiguration.getFetchMaxWait());
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, kafkaConsumerConfiguration.getAutoCommitIntervalMsConfig());
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, kafkaConsumerConfiguration.getSessionTimeoutMsConfig());
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, kafkaConsumerConfiguration.getMaxPollRecordsConfig());
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, kafkaConsumerConfiguration.getMaxPollIntervalConfig());
        //key序列化器选择
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        //value序列化器选择
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        //设置sasl认证
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
        props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
        props.put(SaslConfigs.SASL_JAAS_CONFIG, "org.apache.kafka.common.security.plain.PlainLoginModule required username='" + kafkaConfiguration.getUserName() + "' password='" + kafkaConfiguration.getPassword() + "';");
        return props;
    }
 
    /**
     * kafka消费者工厂
     */
    @Bean
    public ConsumerFactory<Object, Object> consumerFactory() {
        return new DefaultKafkaConsumerFactory(consumerConfigs());
    }
 
 
    /**
     * 监听工厂
     */
    @Bean
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Object, Object>> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        //线程数
        factory.setConcurrency(kafkaListenerConfiguration.getConcurrency());
        //手动提交
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        //开启批量处理
        factory.setBatchListener(kafkaListenerConfiguration.getBatchListener());
        factory.getContainerProperties().setPollTimeout(kafkaListenerConfiguration.getPollTimeout());
        return factory;
    }
 
}

ProducerFactoryBuilder 生产者工厂

 
@Configuration
public class ProducerFactoryBuilder {
 
    @Autowired
    private KafkaConfiguration kafkaConfiguration;
 
    @Autowired
    private KafkaProducerConfiguration kafkaProducerConfiguration;
 
 
    /**
     * 生产者配置
     *
     * @return 配置
     */
    @Bean
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>(11);
        //kafka server地址
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConfiguration.getBootstrapServers());
        /*
         * acks = 0 如果设置为零，则生产者将不会等待来自服务器的任何确认，该记录将立即添加到套接字缓冲区并视为已发送。在这种情况下，无法保证服务器已收到记录，并且重试配置将不会生效（因为客户端通常不会知道任何故障），为每条记录返回的偏移量始终设置为 - 1。
         * acks = 1 这意味着leader会将记录写入其本地日志，但无需等待所有副本服务器的完全确认即可做出回应，在这种情况下，如果leader在确认记录后立即失败，但在将数据复制到所有的副本服务器之前，则记录将会丢失。
         * acks = all 这意味着leader将等待完整的同步副本集以确认记录，这保证了只要至少一个同步副本服务器仍然存活，记录就不会丢失，这是最强有力的保证，这相当于acks = -1 的设置。
         */
        props.put(ProducerConfig.ACKS_CONFIG, kafkaProducerConfiguration.getAcks());
        //消息发送失败重试次数
        props.put(ProducerConfig.RETRIES_CONFIG, kafkaProducerConfiguration.getRetries());
        //去缓冲区中一次拉16k的数据，发送到broker
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, kafkaProducerConfiguration.getBatchSize());
        // 批量发送，延迟为30毫秒，如果30ms内凑不够batch则强制发送，提高并发
        props.put(ProducerConfig.LINGER_MS_CONFIG, kafkaProducerConfiguration.getLingerMs());
        //设置缓存区大小 32m
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, kafkaProducerConfiguration.getBufferMemory());
        //key序列化器选择
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        //value序列化器选择
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        //设置sasl认证
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
        props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
        props.put(SaslConfigs.SASL_JAAS_CONFIG, "org.apache.kafka.common.security.plain.PlainLoginModule required username='" + kafkaConfiguration.getUserName() + "' password='" + kafkaConfiguration.getPassword() + "';");
        return props;
    }
 
 
    /**
     * Producer Template 配置
     */
    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        Map<String, Object> stringObjectMap = producerConfigs();
        DefaultKafkaProducerFactory<String, String> objectObjectDefaultKafkaProducerFactory = new DefaultKafkaProducerFactory<>(stringObjectMap);
        return new KafkaTemplate<>(objectObjectDefaultKafkaProducerFactory);
    }
}

配置文件

kafka:
server-host: 192.168.1.113:9048

使用示例

@Slf4j
@RestController
@RequestMapping("test")
public class TestController {
 
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
 
 
    @PostMapping("test")
    public RestResult test() {
        kafkaTemplate.send("TOPIC_NAME", 0, "key", "this is a message");
        return RestResult.wrapSuccessResponse();
    }
 
    @KafkaListener(topics = "TOPIC_NAME", groupId = "MyGroup1", containerFactory = "kafkaListenerContainerFactory")
    public void kafkaListener(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) {
        for (ConsumerRecord item : records) {
            System.out.printf("topic is %s, offset is %d,partition is %s, value is %s \n", item.topic(), item.offset(), item.partition(), item.value());
            log.info("topic is : {}, offset is : {},partition is : {}, value is : {}",item.topic(), item.offset(), item.partition(), item.value());
        }
        ack.acknowledge();
    }
 
 
}

Kafka开源项目指南

Kafka开源项目指南提供详尽教程，助开发者掌握其架构、配置和使用，实现高效数据流管理和实时处理。它高性能、可扩展，适合日志收集和实时数据处理，通过持久化保障数据安全，是企业大数据生态系统的核心。

更多推荐

Kafka入门（一）概述、部署与API的简单使用

Kafka概述、部署与API的简单使用

Kafka开源项目指南

基于canal和kafka同步，实现binlog同步ElasticSearch

文章目录前言elasticsearch 安装canal安装canal-adapter 安装及配置mysql 安装zk及kafaka安装查看效果注意事项前言中间件版本elasticsearch7.5.2canal1.1.4client-adapter1.1.5-alpha-1zookeeper3.4.13kafka2.6.0mysql5.7.31elasticsearch 安装{"settings"

Kafka开源项目指南

基于 Iceberg 的湖仓一体架构在 B 站的实践

背景在B站，每天都有PB级的数据注入到大数据平台，经过离线或实时的ETL建模后，提供给下游的分析、推荐及预测等场景使用。面对如此大规模的数据，如何高效低成本地满足下游数据的分析需求，一直是我们重点的工作方向。我们之前的数据处理流程基本上是这样的：采集端将客户端埋点、服务端埋点、日志、业务数据库等数据收集到HDFS、Kafka等存储系统中，然后通过Hive、Spark、Fl...