kafka实战文章目录
` 提示：kafka在java代码的最简单应用
前言
提示：这里可以添加本文要记录的大概内容：
例如：随着人工智能的不断发展，机器学习这门技术也越来越重要，很多人都开启了学习机器学习，本文就介绍了机器学习的基础内容。
提示：以下是本篇文章正文内容，下面案例可供参考
一、引入jar包
<!-- kafka -->
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
二、使用步骤
1.配置kafka基础信息
代码如下（示例）：
spring:
kafka:
# 指定kafka server的地址，集群配多个，中间，逗号隔开
bootstrap-servers: ip:9092
producer:
#重试次数
retries: 3
#批量发送的消息数量
batch-size: 1000
#32MB的批处理缓冲区
buffer-memory: 33554432
consumer:
#默认消费者组
group-id: dcp-group-XXX
#批量一次最大拉取数据量
max-poll-records: 4000
#是否自动提交
enable-auto-commit: false
#自动提交时间间隔，单位ms
#auto-commit-interval: 1000
#最早未被消费的offset
auto-offset-reset: earliest
# earliest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
# latest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
# none:topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
listener:
concurrency: 3
ack-mode: MANUAL_IMMEDIATE
# 消费者监听的topic不存在时，项目会报错，设置为false
missing-topics-fatal: false
type: batch
2.基础配置类
代码如下（示例）：
package com.zmj.tl.kafka.kafka.config;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.listener.ContainerProperties;
import java.util.HashMap;
import java.util.Map;
@Configuration
public class KafkaConfiguration {
@Value("${spring.kafka.bootstrap-servers}")
private String bootstrapServers;
@Value("${spring.kafka.producer.retries}")
private Integer retries;
@Value("${spring.kafka.producer.batch-size}")
private Integer batchSize;
@Value("${spring.kafka.producer.buffer-memory}")
private Integer bufferMemory;
@Value("${spring.kafka.consumer.group-id}")
private String groupId;
@Value("${spring.kafka.consumer.auto-offset-reset}")
private String autoOffsetReset;
@Value("${spring.kafka.consumer.max-poll-records}")
private Integer maxPollRecords;
@Value("${spring.kafka.listener.concurrency}")
private Integer batchConcurrency;
@Value("${spring.kafka.consumer.enable-auto-commit}")
private Boolean autoCommit;
/**
* 生产者配置信息
*/
@Bean
public Map<String, Object> producerConfigs() {
Map<String, Object> props = new HashMap<String, Object>();
props.put(ProducerConfig.ACKS_CONFIG, "0");
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
props.put(ProducerConfig.RETRIES_CONFIG, retries);
props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
return props;
}
/**
* 生产者工厂
*/
@Bean
public ProducerFactory<String, String> producerFactory() {
return new DefaultKafkaProducerFactory<String, String>(producerConfigs());
}
/**
* 生产者模板
*/
@Bean
public KafkaTemplate<String, String> kafkaTemplate() {
return new KafkaTemplate<>(producerFactory());
}
/**
* 消费者配置信息
*/
@Bean
public Map<String, Object> consumerConfigs() {
Map<String, Object> props = new HashMap<String, Object>();
props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, autoCommit);
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 6000);
props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 2000);
props.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);
props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 60000);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
return props;
}
/**
* 消费者批量工厂
*/
@Bean("batchFactory")
public KafkaListenerContainerFactory<?> batchFactory() {
ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerConfigs()));
//设置并发量，小于或等于Topic的分区数
factory.setConcurrency(batchConcurrency);
factory.getContainerProperties().setPollTimeout(1500);
factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
//设置为批量消费，每个批次数量在Kafka配置参数中设置ConsumerConfig.MAX_POLL_RECORDS_CONFIG
factory.setBatchListener(true);
return factory;
}
}
此处初始化kafka的一些基础配置
3.发送kafka消息
代码如下（示例）：
/**
* 注入bean
*/
private final static KafkaTemplate<String, String> KAFKATEMOLATE = SpringUtil.getBean(KafkaTemplate.class);
public static void sendMessage(String topic,Object message) {
KAFKATEMOLATE.send(topic, message.toString());
}
3.监听kafka消息
@KafkaListener(topics = {
TOPIC_3}, groupId = "aaa" )
private void topic3(List<ConsumerRecord<?, ?>> consumerRecords, Acknowledgment ack) {
for (ConsumerRecord<?, ?> consumerRecord : consumerRecords) {
System.out.println("TOPIC_3:"+String.format("key:%s , value:%s , offset:%s", consumerRecord.key(), consumerRecord.value(), consumerRecord.offset()));
}
}
可直接使用@KafkaListener注解进行kafka消息的监听，需要注意的是如果是没有设置自动提交offset,则需要每次消费完手动提交及如下代码
@KafkaListener(topics = {
TOPIC_ONE},groupId = "aaa" )
private void topic(List<ConsumerRecord<?, ?>> consumerRecords, Acknowledgment ack) {
for (ConsumerRecord<?, ?> consumerRecord : consumerRecords) {
System.out.println("TOPIC_ONE:"+String.format("key:%s , value:%s , offset:%s", consumerRecord.key(), consumerRecord.value(), consumerRecord.offset()));
}
ack.acknowledge();
}
ack.acknowledge();就是手动提交的方法
需要注意的是：如果写了消费组则组内消息只会消费一次，如果没有多个订阅者都没有配置消费组组则他们都属于一个默认组，也只会有一个消费者消费到消息
总结
以上就是对Kafka在代码中的简单的使用，希望能帮助到大家
