blog/spring-samples-for-all
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

增加 kafka 整合用例

Browse Source
This commit is contained in:
罗祥
2019-01-09 14:45:27 +08:00
parent fc0c2b90b9
commit 7f735f9317
14 changed files with 495 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
spring-boot/spring-boot-kafka/src/main/java/com/heibaiying/springboot/Producer/KafKaCustomrProducer.java
View File

@ -28,6 +28,7 @@ public class KafKaCustomrProducer {
* 被用于ProducerRecord<K, V> producerRecord,即生产者发送消息的key,value 类型
*/
ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic, object);
future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
@Override
public void onFailure(Throwable throwable) {

21
spring-boot/spring-boot-kafka/src/main/java/com/heibaiying/springboot/config/KafkaConfig.java Normal file
View File

@ -0,0 +1,21 @@
package com.heibaiying.springboot.config;
import com.heibaiying.springboot.constant.Topic;
import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* @author : heibaiying
* @description : kafka配置类
*/
@Configuration
public class KafkaConfig {
@Bean
public NewTopic groupTopic() {
// 指定主题名称，分区数量，和复制因子
return new NewTopic(Topic.GROUP, 10, (short) 2);
}
}

2
spring-boot/spring-boot-kafka/src/main/java/com/heibaiying/springboot/constant/Topic.java
View File

@ -3,7 +3,7 @@ package com.heibaiying.springboot.constant;
// 实际开发时建议写到配置文件中
public class Topic {
public static final String GROUP="spring.boot.kafka.group";
public static final String GROUP="spring.boot.kafka.newGroup";
public static final String SIMPLE="spring.boot.kafka.simple";
public static final String BEAN="spring.boot.kafka.bean";

2
spring-boot/spring-boot-kafka/src/main/java/com/heibaiying/springboot/consumer/KafkaBeanConsumer.java
View File

@ -17,7 +17,7 @@ import org.springframework.stereotype.Component;
@Slf4j
public class KafkaBeanConsumer {
@KafkaListener(groupId = "beanGroup", topics = Topic.BEAN)
@KafkaListener(groupId = "beanGroup",topics = Topic.BEAN)
public void consumer(ConsumerRecord<String, Object> record) {
System.out.println("消费者收到消息:" + JSON.parseObject(record.value().toString(), Programmer.class));
}

22
spring-boot/spring-boot-kafka/src/main/java/com/heibaiying/springboot/consumer/KafkaGroupConsumer.java
View File

@ -4,34 +4,46 @@ import com.heibaiying.springboot.constant.Topic;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.stereotype.Component;
/**
* @author : heibaiying
* @description : kafka 消费者组
*
* <p>
* 多个消费者群组可以共同读取同一个主题，彼此之间互不影响。
* 但主题的一个分区只能被同一个消费者群组里面的一个消费者读取。
*/
@Component
@Slf4j
public class KafkaGroupConsumer {
// 分组1 中的消费者1
@KafkaListener(id = "consumer1-1", groupId = "group1", topics = Topic.GROUP)
@KafkaListener(id = "consumer1-1", groupId = "group1", topicPartitions =
{@TopicPartition(topic = Topic.GROUP, partitions = {"0", "1"})
})
public void consumer1_1(ConsumerRecord<String, Object> record) {
System.out.println("consumer1-1 收到消息:" + record.value());
}
// 分组1 中的消费者2
@KafkaListener(id = "consumer1-2", groupId = "group1", topics = Topic.GROUP)
@KafkaListener(id = "consumer1-2", groupId = "group1", topicPartitions =
{@TopicPartition(topic = Topic.GROUP, partitions = {"2", "3"})
})
public void consumer1_2(ConsumerRecord<String, Object> record) {
System.out.println("consumer1-2 收到消息:" + record.value());
}
// 分组1 中的消费者3
@KafkaListener(id = "consumer1-3", groupId = "group1", topicPartitions =
{@TopicPartition(topic = Topic.GROUP, partitions = {"0", "1"})
})
public void consumer1_3(ConsumerRecord<String, Object> record) {
System.out.println("consumer1-3 收到消息:" + record.value());
}
// 分组2 中的消费者
@KafkaListener(id = "consumer2-1", groupId = "group2", topics = Topic.GROUP)
public void consumer2_1(ConsumerRecord<String, Object> record) {
System.out.println("consumer2-1 收到消息:" + record.value());
System.err.println("consumer2-1 收到消息:" + record.value());
}
}

13
spring-boot/spring-boot-kafka/src/main/java/com/heibaiying/springboot/consumer/KafkaSimpleConsumer.java
View File

@ -4,11 +4,16 @@ import com.heibaiying.springboot.constant.Topic;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.PartitionInfo;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
import java.util.List;
import java.util.Map;
/**
* @author : heibaiying
* @description : kafka 简单消息消费者
@ -20,8 +25,12 @@ public class KafkaSimpleConsumer {
// 简单消费者
@KafkaListener(groupId = "simpleGroup", topics = Topic.SIMPLE)
public void consumer1_1(ConsumerRecord<String, Object> record, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,Consumer consumer) {
public void consumer1_1(ConsumerRecord<String, Object> record, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic, Consumer consumer) {
System.out.println("消费者收到消息:" + record.value() + "; topic:" + topic);
consumer.commitSync();
/*
* 如果需要手工提交异步 consumer.commitSync();
* 手工同步提交 consumer.commitAsync()
*/
}
}

70
spring-boot/spring-boot-kafka/src/main/java/com/heibaiying/springboot/controller/SendMsgController.java Normal file
View File

@ -0,0 +1,70 @@
package com.heibaiying.springboot.controller;
import com.alibaba.fastjson.JSON;
import com.heibaiying.springboot.Producer.KafKaCustomrProducer;
import com.heibaiying.springboot.bean.Programmer;
import com.heibaiying.springboot.constant.Topic;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.Date;
/**
* @author : heibaiying
* @description : 测试消息发送
*/
@Slf4j
@RestController
public class SendMsgController {
@Autowired
private KafKaCustomrProducer producer;
@Autowired
private KafkaTemplate kafkaTemplate;
/***
* 发送消息体为基本类型的消息
*/
@GetMapping("sendSimple")
public void sendSimple() {
producer.sendMessage(Topic.SIMPLE, "hello spring boot kafka");
}
/***
* 发送消息体为bean的消息
*/
@GetMapping("sendBean")
public void sendBean() {
Programmer programmer = new Programmer("xiaoming", 12, 21212.33f, new Date());
producer.sendMessage(Topic.BEAN, JSON.toJSON(programmer).toString());
}
/***
* 多消费者组、组中多消费者对同一主题的消费情况
*/
@GetMapping("sendGroup")
public void sendGroup() {
for (int i = 0; i < 4; i++) {
ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(Topic.GROUP, i % 4, "key", "hello group " + i);
future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
@Override
public void onFailure(Throwable throwable) {
log.info("发送消息失败:" + throwable.getMessage());
}
@Override
public void onSuccess(SendResult<String, Object> sendResult) {
System.out.println("发送结果:" + sendResult.toString());
}
});
}
}
}

2
spring-boot/spring-boot-kafka/src/main/resources/application.yml
View File

@ -25,7 +25,7 @@ spring:
# earliest ：在偏移量无效的情况下，消费者将从起始位置读取分区的记录
auto-offset-reset: earliest
# 是否自动提交偏移量，默认值是true,为了避免出现重复数据和数据丢失，可以把它设置为false,然后手动提交偏移量
enable-auto-commit: false
enable-auto-commit: true
# 键的反序列化方式
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
# 值的反序列化方式

51
spring-boot/spring-boot-kafka/src/test/java/com/heibaiying/springboot/KafkaTests.java
View File

@ -1,51 +0,0 @@
package com.heibaiying.springboot;
import com.alibaba.fastjson.JSON;
import com.heibaiying.springboot.Producer.KafKaCustomrProducer;
import com.heibaiying.springboot.bean.Programmer;
import com.heibaiying.springboot.constant.Topic;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.test.context.junit4.SpringRunner;
import java.util.Date;
@RunWith(SpringRunner.class)
@SpringBootTest
public class KafkaTests {
@Autowired
private KafKaCustomrProducer producer;
/***
* 发送消息体为基本类型的消息
*/
@Test
public void sendSimple() {
producer.sendMessage(Topic.SIMPLE, "hello spring boot kafka");
}
/***
* 发送消息体为bean的消息
*/
@Test
public void sendBean() {
Programmer programmer = new Programmer("xiaoming", 12, 21212.33f, new Date());
producer.sendMessage(Topic.BEAN, JSON.toJSON(programmer).toString());
}
/***
* 多消费者组、组中多消费者对同一主题的消费情况
*/
@Test
public void sendGroup() {
for (int i = 0; i < 5; i++) {
producer.sendMessage(Topic.GROUP, "hello group " + i);
}
}
}