在上一篇文章:SpringBoot中Kafka的SSL链接_长河的博客-CSDN博客_kafka ssl连接中我们配置了kafka的SSL的生产者的配置, 现在配置其消费者:

首先是测试代码:

private static void testConusmer() throws Exception {

    String truststorePath = "./" + UUID.randomUUID().toString().replace("-", "");
    FileToBase64Tools.decoderBase64File(SSL_TRUSTSTORE_LOCATION, truststorePath);

    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS_CONFIG);

    props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, truststorePath);
    String jaasConfig = "org.apache.kafka.common.security.scram.ScramLoginModule required username='td' password='v';";
    props.put(AdminClientConfig.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
    props.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-512");
    props.put(SaslConfigs.SASL_JAAS_CONFIG, jaasConfig);
    props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, truststorePath);
    props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "R");

    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
    props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000);
    props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100);
    props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 120000);
    props.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, 120000);
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
    DefaultKafkaConsumerFactory<Object, Object> factory = new DefaultKafkaConsumerFactory<>(props);
    System.out.println(factory);
}

再配置Spring管理的bean:

@Bean
public ConsumerFactory<String, String> consumerFactory() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);

    String truststorePath = "./" + UUID.randomUUID().toString().replace("-", "");
    try {
        FileToBase64Tools.decoderBase64File(SSL_TRUSTSTORE_LOCATION, truststorePath);
        props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, truststorePath);
    } catch (Exception e) {
        log.error("decode Kafka SSL trust-store-location file occurred an error: ", e);
    }

    String jaasConfig = "org.apache.kafka.common.security.scram.ScramLoginModule required username='td' password='v';";
    props.put(AdminClientConfig.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
    props.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-512");
    props.put(SaslConfigs.SASL_JAAS_CONFIG, jaasConfig);
    props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "R");

    props.put("group.id", "test");
    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
    props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000);
    props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100);
    props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 120000);
    props.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, 120000);
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
    return new DefaultKafkaConsumerFactory<>(props);
}

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setBatchListener(false);
    factory.setConcurrency(1);
    return factory;
}

注意点:  jaasConfig中的用户名和密码都必要是单引号, 否则链接会报错,

在写一个事件的消费者:

package com;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
@Component
public class TestKafkaConsumer {

    private static AtomicInteger index = new AtomicInteger(1);

    @KafkaListener(id = "testcsd1", topics = {"messages.test"})
    public void testTheConsumer(ConsumerRecord<String, Object> record) {
        log.info("now is {},the message is:{}, the index is:{}",
                LocalDateTime.now(), record, index.getAndIncrement());
    }

}