SpringBoot集成kafka全⾯实战
本⽂是SpringBoot+Kafka的实战讲解，如果对kafka的架构原理还不了解的读者，建议先看⼀下《⼤⽩话kafka架构原理》、《秒懂kafka HA（⾼可⽤）》两篇⽂章。
⼀、⽣产者实践
普通⽣产者
带回调的⽣产者
⾃定义分区器
kafka事务提交
⼆、消费者实践
简单消费
指定topic、partition、offset消费
批量消费
监听异常处理器
消息过滤器
消息转发
定时启动/停⽌
⼀、前戏
1、在项⽬中连接kafka，因为是外⽹，⾸先要开放kafka配置⽂件中的如下配置（其中IP为公⽹IP），
advertised.listeners=PLAINTEXT://112.126.74.249:9092
2、在开始前我们先创建两个topic：topic1、topic2，其分区和副本数都设置为2，⽤来测试，
[root@iZ2zegzlkedbo3e64vkbefZ ~]# cd /usr/local/kafka-cluster/kafka1/bin/
[root@iZ2zegzlkedbo3e64vkbefZ bin]# ./kafka-topics.sh --create --zookeeper 172.17.80.219:2181 --replication-factor 2 --partitions 2 --topic topic1
Created topic topic1.
[root@iZ2zegzlkedbo3e64vkbefZ bin]# ./kafka-topics.sh --create --zookeeper 172.17.80.219:2181 --replication-factor 2 --partitions 2 --topic topic2
Created topic topic2.
当然我们也可以不⼿动创建topic，在执⾏代码kafkaTemplate.send("topic1", normalMessage)发送消息时，kafka会帮我们⾃动完成topic的创建⼯作，但这种情况下创建的topic默认只有⼀个分区，分区也没有副本。所以，我们可以在项⽬中新建⼀个配置类专门⽤来初始化topic，如下，
@Configuration
public class KafkaInitialConfiguration {
// 创建⼀个名为testtopic的Topic并设置分区数为8，分区副本数为2
@Bean
public NewTopic initialTopic() {
return new NewTopic("testtopic",8, (short) 2 );
}
// 如果要修改分区数，只需修改配置值重启项⽬即可
// 修改分区数并不会导致数据的丢失，但是分区数只能增⼤不能减⼩
@Bean
public NewTopic updateTopic() {
return new NewTopic("testtopic",10, (short) 2 );
}
}
3、新建SpringBoot项⽬
①引⼊pom依赖
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
② application.propertise配置（本⽂⽤到的配置项这⾥全列了出来）
>>#【Kafka集】>>#
spring.kafka.bootstrap-servers=112.126.74.249:9092,112.126.74.249:9093
>>#【初始化⽣产者配置】>>#
# 重试次数
spring.ies=0
# 应答级别：多少个分区副本备份完成时向⽣产者发送ack确认(可选0、1、all/-1)
spring.kafka.producer.acks=1
# 批量⼤⼩
spring.kafka.producer.batch-size=16384
# 提交延时
spring.kafka.producer.properties.linger.ms=0
# 当⽣产端积累的消息达到batch-size或接收到消息linger.ms后,⽣产者就会将消息提交给kafka
# linger.ms为0表⽰每接收到⼀条消息就提交给kafka，这时候batch-size其实就没⽤了
# ⽣产端缓冲区⼤⼩
spring.kafka.producer.buffer-memory = 33554432
# Kafka提供的序列化和反序列化类
spring.kafka.producer.key-serializer=org.apache.kafkamon.serialization.StringSerializer spring.kafka.producer.value-serializer=org.apache.kafkamon.serialization.StringSerializer
# ⾃定义分区器
# spring.kafka.producer.properties.partitioner.class=com.felix.kafka.producer.CustomizePartitioner >>#【初始化消费者配置】>>#
# 默认的消费组ID
up.id=defaultConsumerGroup
# 是否⾃动提交offset
able-auto-commit=true
# 提交offset延时(接收到消息后多久提交offset)
sumer.automit.interval.ms=1000
# 当kafka中没有初始offset或offset超出范围时将⾃动重置offset
# earliest：重置为分区中最⼩的offset;
# latest：重置为分区中最新的offset(消费分区中新产⽣的数据);
# none：只要有⼀个分区不存在已提交的offset，就抛出异常;
sumer.auto-offset-reset=latest
# 消费会话超时时间(超过这个时间consumer没有发送⼼跳，就会触发rebalance操作)
sumer.properties.session.timeout.ms=120000
# 消费请求超时时间
quest.timeout.ms=180000
# Kafka提供的序列化和反序列化类
sumer.key-deserializer=org.apache.kafkamon.serialization.StringDeserializer sumer.value-deserializer=org.apache.kafkamon.serialization.StringDeserializer # 消费端监听的topic不存在时，项⽬启动会报错(关掉)
spring.kafka.listener.missing-topics-fatal=false
# 设置批量消费
# spring.pe=batch
# 批量消费每次最多消费多少条消息
# sumer.max-poll-records=50
⼆、Hello Kafka
1、简单⽣产者
@RestController
public class KafkaProducer {
@Autowired
private KafkaTemplate<String, Object> kafkaTemplate;
// 发送消息
@GetMapping("/kafka/normal/{message}")
public void sendMessage1(@PathVariable("message") String normalMessage) { kafkaTemplate.send("topic1", normalMessage);
}
}
2、简单消费
@Component
public class KafkaConsumer {springboot原理是什么
// 消费监听
@KafkaListener(topics = {"topic1"})
public void onMessage1(ConsumerRecord<?, ?> record){
// 消费的哪个topic、partition的消息，打印出消息内容
System.out.println("简单消费："+pic()+"-"+record.partition()+"-"+record.value());
}
}
上⾯⽰例创建了⼀个⽣产者，发送消息到topic1，消费者监听topic1消费消息。⽤@KafkaListener注解，topics表⽰监听的topic，⽀持同时监听多个，⽤英⽂逗号分隔。启动项⽬，postman调接⼝触发⽣产者发送消息，
可以看到消费成功，
三、⽣产者
1、带回调的⽣产者
kafkaTemplate提供了⼀个回调⽅法addCallback，我们可以在回调⽅法中监控消息是否发送成功或失败时做补偿处理，有两种写法，
@GetMapping("/kafka/callbackOne/{message}")
public void sendMessage2(@PathVariable("message") String callbackMessage) {
kafkaTemplate.send("topic1", callbackMessage).addCallback(success -> {
/
/ 消息发送到的topic
String topic = RecordMetadata().topic();
// 消息发送到的分区
int partition = RecordMetadata().partition();
// 消息在分区内的offset
long offset = RecordMetadata().offset();
System.out.println("发送消息成功:" + topic + "-" + partition + "-" + offset);
}, failure -> {
System.out.println("发送消息失败:" + Message());
});
}
@GetMapping("/kafka/callbackTwo/{message}")
public void sendMessage3(@PathVariable("message") String callbackMessage) {
kafkaTemplate.send("topic1", callbackMessage).addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
@Override
public void onFailure(Throwable ex) {
System.out.println("发送消息失败："+ex.getMessage());
}
@Override
public void onSuccess(SendResult<String, Object> result) {
System.out.println("发送消息成功：" + RecordMetadata().topic() + "-"
+ RecordMetadata().partition() + "-" + RecordMetadata().offset());
}
});
}
2、⾃定义分区器
我们知道，kafka中每个topic被划分为多个分区，那么⽣产者将消息发送到topic时，具体追加到哪个分区呢？这就是所谓的分区策略，Kafka 为我们提供了默认的分区策略，同时它也⽀持⾃定义分区策略。其路由机制为：
①若发送消息时指定了分区（即⾃定义分区策略），则直接将消息append到指定分区；
②若发送消息时未指定 patition，但指定了 key（kafka允许为每条消息设置⼀个key），则对key值进⾏hash计算，根据计算结果路由到指定分区，这种情况下可以保证同⼀个 Key 的所有消息都进⼊到相同的分区；
③ patition 和 key 都未指定，则使⽤kafka默认的分区策略，轮询选出⼀个 patition；
※我们来⾃定义⼀个分区策略，将消息发送到我们指定的partition，⾸先新建⼀个分区器类实现Partitioner接⼝，重写⽅法，其中partition⽅法的返回值就表⽰将消息发送到⼏号分区，
public class CustomizePartitioner implements Partitioner {
@Override
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
// ⾃定义分区规则(这⾥假设全部发到0号分区)
// ......
return 0;
}
@Override
public void close() {
}
@Override
public void configure(Map<String, ?> configs) {
}
}
在application.propertise中配置⾃定义分区器，配置的值就是分区器类的全路径名，
# ⾃定义分区器
spring.kafka.producer.properties.partitioner.class=com.felix.kafka.producer.CustomizePartitioner
3、kafka事务提交
如果在发送消息时需要创建事务，可以使⽤ KafkaTemplate 的 executeInTransaction ⽅法来声明事务，
@GetMapping("/kafka/transaction")
public void sendMessage7(){
// 声明事务：后⾯报错消息不会发出去
operations.send("topic1","test executeInTransaction");
throw new RuntimeException("fail");
});
// 不声明事务：后⾯报错但前⾯消息已经发送成功了
kafkaTemplate.send("topic1","test executeInTransaction");
throw new RuntimeException("fail");
}
四、消费者
1、指定topic、partition、offset消费
前⾯我们在监听消费topic1的时候，监听的是topic1上所有的消息，如果我们想指定topic、指定partition、指定offset来消费呢？也很简单，@KafkaListener注解已全部为我们提供，
/**
* @Title 指定topic、partition、offset消费
* @Description 同时监听topic1和topic2，监听topic1的0号分区、topic2的 "0号和1号" 分区，指向1号分区的offset初始值为8
* @Author long.yuan
* @Date 2020/3/22 13:38
* @Param [record]
* @return void
**/
@KafkaListener(id = "consumer1",groupId = "felix-group",topicPartitions = {
@TopicPartition(topic = "topic1", partitions = { "0" }),
@TopicPartition(topic = "topic2", partitions = "0", partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "8"))
})
public void onMessage2(ConsumerRecord<?, ?> record) {
System.out.println("topic:"+pic()+"|partition:"+record.partition()+"|offset:"+record.offset()+"|value:"+record.value());
}
属性解释：
① id：消费者ID；
② groupId：消费组ID；
③ topics：监听的topic，可监听多个；
④ topicPartitions：可配置更加详细的监听信息，可指定topic、parition、offset监听。
上⾯onMessage2监听的含义：监听topic1的0号分区，同时监听topic2的0号分区和topic2的1号分区⾥⾯offset从8开始的消息。
注意：topics和topicPartitions不能同时使⽤；
2、批量消费
设置application.prpertise开启批量消费即可，
# 设置批量消费
spring.pe=batch
# 批量消费每次最多消费多少条消息
sumer.max-poll-records=50
接收消息时⽤List来接收，监听代码如下，
@KafkaListener(id = "consumer2",groupId = "felix-group", topics = "topic1")
public void onMessage3(List<ConsumerRecord<?, ?>> records) {
System.out.println(">>>批量消费⼀次，records.size()="+records.size());
for (ConsumerRecord<?, ?> record : records) {
System.out.println(record.value());
}
}
3、ConsumerAwareListenerErrorHandler 异常处理器
通过异常处理器，我们可以处理consumer在消费时发⽣的异常。
新建⼀个 ConsumerAwareListenerErrorHandler 类型的异常处理⽅法，⽤@Bean注⼊，BeanName默认就是⽅法名，然后我们将这个异常处理器的BeanName放到@KafkaListener注解的errorHandler属性⾥⾯，当监听抛出异常的时候，则会⾃动调⽤异常处理器，
// 新建⼀个异常处理器，⽤@Bean注⼊
@Bean
public ConsumerAwareListenerErrorHandler consumerAwareErrorHandler() {
return (message, exception, consumer) -> {
System.out.println("消费异常："+Payload());
return null;
};
}
// 将这个异常处理器的BeanName放到@KafkaListener注解的errorHandler属性⾥⾯
@KafkaListener(topics = {"topic1"},errorHandler = "consumerAwareErrorHandler")
public void onMessage4(ConsumerRecord<?, ?> record) throws Exception {
throw new Exception("简单消费-模拟异常");
}
// 批量消费也⼀样，异常处理器的Payload()也可以拿到各条消息的信息
@KafkaListener(topics = "topic1",errorHandler="consumerAwareErrorHandler")
public void onMessage5(List<ConsumerRecord<?, ?>> records) throws Exception {
System.out.println("批量消费⼀次...");
throw new Exception("批量消费-模拟异常");
}
执⾏看⼀下效果，
4、消息过滤器
消息过滤器可以在消息抵达consumer之前被拦截，在实际应⽤中，我们可以根据⾃⼰的业务逻辑，筛选出需要的信息再交由KafkaListener 处理，不需要的消息则过滤掉。
配置消息过滤只需要为⼯⼚配置⼀个RecordFilterStrategy（消息过滤策略），返回true的时候消息将会被抛弃，返回false时，消息能正常抵达监听容器。
@Component
public class KafkaConsumer {
@Autowired
ConsumerFactory consumerFactory;
// 消息过滤器
@Bean
public ConcurrentKafkaListenerContainerFactory filterContainerFactory() {
ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
factory.setConsumerFactory(consumerFactory);
// 被过滤的消息将被丢弃
factory.setAckDiscarded(true);
// 消息过滤策略
factory.setRecordFilterStrategy(consumerRecord -> {
if (Integer.parseInt(consumerRecord.value().toString()) % 2 == 0) {
return false;
}

版权声明：本站内容均来自互联网，仅供演示用，请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系QQ:729038198，我们将在24小时内删除。

maven 原理
« 上一篇
android robust 原理
下一篇 »

发表评论