引言
Apache Kafka 是一个分布式流处理平台，广泛应用于构建实时数据管道和流应用。随着数据敏感性的增加，安全成为Kafka部署中的一个重要考虑因素。SASL（Simple Authentication and Security Layer）是一种为网络协议提供身份验证机制的框架。本文将详细介绍如何在Kafka中配置SASL认证，以增强集群的安全性。
kafka配置
1.在config/server.properties中配置如下信息
listeners=SASL_PLAINTEXT://192.168.2.122:9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN
super.users=User:admin
2.配置一个名 kafka_server_jaas.conf 和 kafka_client_jaas.conf 的配置文件，将配置文件放置在config目录下
备注：user_来定义多个用户，供客户端程序（生产者、消费者程序）认证使用
KafkaServer {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="admin"
password="admin"
user_admin="admin"
user_yianweilai="yian@2024";
};
KafkaClient {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="yianweilai"
password="yian@2024"
};
3.修改脚本
export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G -Djava.security.auth.login.config=/usr/local/kafka/config/kafka_server_jaas.conf"
4.添加相应的生产者和消费者认证文件路径
export KAFKA_OPTS="-Djava.security.auth.login.config=/usr/local/kafka/config/kafka_client_jaas.conf"
客户端配置
kafka:
# 堆代码 duidaima.com
# kafka 服务器地址
server-host: 192.168.2.122:9092
properties:
sasl:
# sasl 认证账号
username: yianweilai
# sasl 密码
password: yian@2024
# 消费者
consumer:
properties:
# 消费组
group-id: default-group
# 消费者默认等待服务响应时间(毫秒)
fetch-max-wait: 5000
# 拉取最大条数
max-poll-record: 100
# 拉取最大间隔
max-poll-interval-ms: 100
# 是否开启自动提交
enable-auto-commit: true
# 消费策略
auto-offset-reset: earliest
# 提交间隔
auto-commit-interval: 1000
# 会话超时时间
session-timeout: 30000
# 生产者
producer:
# 重试次数
retries: 0
# 确认机制
acks: all
# 批量大小
batch-size: 16384
# 延迟时间
linger.ms: 30
# 缓冲大小
buffer-memory: 33554432
# 监听
listener:
# 并发度
concurrency: 3
# 确认机制
ack-mode: manual_immediate
# 拉取最大时间
poll-timeout: 3000
# 是否开启批量处理
batch_listener: true
公共配置
@Data
@Configuration
public class KafkaConfiguration {
/**
* 主机地址
*/
@Value("${kafka.server-host:}")
private String bootstrapServers;
/**
* sasl 认证账号
*/
@Value("${kafka.properties.sasl.username:admin}")
private String userName;
/**
* sasl 密码
*/
@Value("${kafka.properties.sasl.password:yian@2024}")
private String password;
}
消费者配置
@Data
@Configuration
public class KafkaConsumerConfiguration {
/**
* 默认组id
*/
@Value("${kafka.consumer.properties.group-id:default-group}")
private String groupId;
@Value("${kafka.consumer.properties.fetch-max-wait:5000}")
private Integer fetchMaxWait;
/**
* 此设置限制每次调用poll返回的消息数，这样可以更容易的预测每次poll间隔要处理的最大值。通过调整此值，可以减少poll间隔，减少重新平衡分组的
*/
@Value("${kafka.consumer.properties.max-poll-record:100}")
private Integer maxPollRecordsConfig;
/**
* 增大poll的间隔，可以为消费者提供更多的时间去处理返回的消息（调用poll（long）返回的消息，通常返回的消息都是一批），缺点是此值越大将会延迟组重新平衡。
*/
@Value("${kafka.consumer.properties.max-poll-interval-ms:100}")
private Integer maxPollIntervalConfig;
/**
* 是否开启自动提交
*/
@Value("${kafka.consumer.properties.enable-auto-commit:#{false}}")
private boolean enableAutoCommitConfig;
/**
* 消费策略
* earliest 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
* latest 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
* none topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
*/
@Value("${kafka.consumer.properties.auto-offset-reset:earliest}")
private String autoOffsetResetConfig;
@Value("${kafka.consumer.properties.auto-commit-interval:1000}")
private String autoCommitIntervalMsConfig;
@Value("${kafka.consumer.properties.session-timeout:30000}")
private String sessionTimeoutMsConfig;
}
生产者配置
@Data
@Configuration
public class KafkaProducerConfiguration {
/**
* 重试次数 默认值0
*/
@Value("${kafka.producer.retries:0}")
private Integer retries;
/**
* acks = 0 如果设置为零，则生产者将不会等待来自服务器的任何确认，该记录将立即添加到套接字缓冲区并视为已发送。在这种情况下，无法保证服务器已收到记录，并且重试配置将不会生效（因为客户端通常不会知道任何故障），为每条记录返回的偏移量始终设置为 - 1。
* acks = 1 这意味着leader会将记录写入其本地日志，但无需等待所有副本服务器的完全确认即可做出回应，在这种情况下，如果leader在确认记录后立即失败，但在将数据复制到所有的副本服务器之前，则记录将会丢失。
* acks = all 这意味着leader将等待完整的同步副本集以确认记录，这保证了只要至少一个同步副本服务器仍然存活，记录就不会丢失，这是最强有力的保证，这相当于acks = -1 的设置。
*/
@Value("${kafka.producer.acks:all}")
private String acks;
/**
* 指定缓存的大小，生产者缓存每个分区未发送的消息。默认 16384
*/
@Value("${kafka.producer.batch-size:16384}")
private Integer batchSize;
/**
* 生产者发送请求之前等待一段时间，设置等待时间是希望更多地消息填补到未满的批中。 默认 30
*/
@Value("${kafka.producer.properties.linger.ms:30}")
private Integer lingerMs;
/**
* 通过KafkaProducer发送出去的消息都是先进入到客户端本地的内存缓冲里，然后把很多消息收集成一个一个的Batch，再发送到Broker上去的 默认32m
*/
@Value("${kafka.producer.buffer-memory:33554432}")
private Integer bufferMemory;
}
监听配置
@Data
@Configuration
public class KafkaListenerConfiguration {
/**
* 启用线程数（提高并发）
*/
@Value("${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("${kafka.listener.ack-mode:manual_immediate}")
private String ackMode;
/**
* 消费超时时间
*/
@Value("${kafka.listener.poll-timeout:3000}")
private Long pollTimeout;
/**
* 是否开启批量处理
*/
@Value("${kafka.listener.batch_listener:#{true}}")
private Boolean batchListener;
}
消费者工厂
@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;
}
}
生产者工厂
@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);
}
}
测试方法
// 监听多个topic
@KafkaListener(topics = {"#{'${kafka.consumer.topic1}'.split(',')}"},group = "MyGroup1",containerFactory = "kafkaListenerContainerFactory")
public void listenPartition(List<ConsumerRecord<?, ?>> records) {
log.info("Id Listener, Thread ID: " + Thread.currentThread().getId());
log.info("Id records size " + records.size());
for (ConsumerRecord<?, ?> record : records) {
Optional<?> kafkaMessage = Optional.ofNullable(record.value());
log.info("Received: " + record);
if (kafkaMessage.isPresent()) {
Object message = record.value();
String topic = record.topic();
int partition = record.partition();
log.info("topic {} partition {} Received message={}", topic,partition,message);
}
}
}
// @KafkaListener(id = "id0", topicPartitions = { @TopicPartition(topic = TPOIC, partitions = { "0" }) })
public void listenPartition0(List<ConsumerRecord<?, ?>> records) {
log.info("Id0 Listener, Thread ID: " + Thread.currentThread().getId());
log.info("Id0 records size " + records.size());
for (ConsumerRecord<?, ?> record : records) {
Optional<?> kafkaMessage = Optional.ofNullable(record.value());
log.info("Received: " + record);
if (kafkaMessage.isPresent()) {
Object message = record.value();
String topic = record.topic();
log.info("topic {} p0 Received message={}", topic,message);
}
}
}
// @KafkaListener(id = "id1", topicPartitions = { @TopicPartition(topic = TPOIC, partitions = { "1" }) })
public void listenPartition1(List<ConsumerRecord<?, ?>> records) {
log.info("Id1 Listener, Thread ID: " + Thread.currentThread().getId());
log.info("Id1 records size " + records.size());
for (ConsumerRecord<?, ?> record : records) {
Optional<?> kafkaMessage = Optional.ofNullable(record.value());
log.info("Received: " + record);
if (kafkaMessage.isPresent()) {
Object message = record.value();
String topic = record.topic();
log.info("topic {} p1 Received message={}", topic,message);
}
}
}
// @KafkaListener(id = "id2", topicPartitions = { @TopicPartition(topic = TPOIC, partitions = { "2" }) })
public void listenPartition2(List<ConsumerRecord<?, ?>> records) {
log.info("Id2 Listener, Thread ID: " + Thread.currentThread().getId());
log.info("Id2 records size " + records.size());
for (ConsumerRecord<?, ?> record : records) {
Optional<?> kafkaMessage = Optional.ofNullable(record.value());
log.info("Received: " + record);
if (kafkaMessage.isPresent()) {
Object message = record.value();
String topic = record.topic();
log.info("topic {} p2 Received message={}", topic, message);
}
}
}
public void send(String topic, String str, int count) {
for (int i=0; i < count; i++) {
template.send(topic, str);
}
}
public void sendJson(String topic, String json, int count) {
for (int i=0; i < count; i++) {
sendJson(topic, json);
}
}
private void sendJson(String topic, String json) {
JSONObject jsonObj = JSON.parseObject(json);
jsonObj.put("topic", topic);
jsonObj.put("ts", System.currentTimeMillis() + "");
ListenableFuture<SendResult<String, String>> future = template.send(topic, jsonObj.toJSONString());
future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
@Override
public void onSuccess(SendResult<String, String> result) {
logger.info("msg OK. " + result.toString());
}
@Override
public void onFailure(Throwable ex) {
logger.error("msg send failed.", ex);
}
});
}
闽公网安备 35020302035485号
