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

新增 spring boot kafka 整合用例

Browse Source
This commit is contained in:
罗祥
2019-01-04 18:00:13 +08:00
parent 8dcfd49cb9
commit df2648e3fd
11 changed files with 335 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
spring-boot/spring-boot-kafka/README.md Normal file
View File

@ -0,0 +1,7 @@
java.time.Duration类下:
private static final Pattern PATTERN =
Pattern.compile("([-+]?)P(?:([-+]?[0-9]+)D)?" +
"(T(?:([-+]?[0-9]+)H)?(?:([-+]?[0-9]+)M)?(?:([-+]?[0-9]+)(?:[.,]([0-9]{0,9}))?S)?)?",
Pattern.CASE_INSENSITIVE);
1s 1.5s 0s 0.001S 1h 2d

62
spring-boot/spring-boot-kafka/pom.xml Normal file
View File

@ -0,0 +1,62 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.1.1.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.heibaiying</groupId>
<artifactId>spring-boot-kafka</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>spring-boot-kafka</name>
<description>kakfa project for Spring Boot</description>
<properties>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.54</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>

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

@ -0,0 +1,45 @@
package com.heibaiying.springboot.Producer;
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.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
/**
* @author : heibaiying
* @description : 生产者
*/
@Component
@Slf4j
public class KafKaCustomrProducer {
@Autowired
private KafkaTemplate kafkaTemplate;
public void sendMessage(String topic, Object object) {
/*
* 这里的ListenableFuture类是spring对java原生Future的扩展增强,是一个泛型接口,用于监听异步方法的回调
* 而对于kafka send 方法返回值而言，这里的泛型所代表的实际类型就是 SendResult<K, V>,而这里K,V的泛型实际上
* 被用于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) {
log.info("发送消息失败:" + throwable.getMessage());
}
@Override
public void onSuccess(SendResult<String, Object> sendResult) {
System.out.println("发送结果:" + sendResult.toString());
}
});
}
}

14
spring-boot/spring-boot-kafka/src/main/java/com/heibaiying/springboot/SpringBootKafkaApplication.java Normal file
View File

@ -0,0 +1,14 @@
package com.heibaiying.springboot;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class SpringBootKafkaApplication {
public static void main(String[] args) {
SpringApplication.run(SpringBootKafkaApplication.class, args);
}
}

24
spring-boot/spring-boot-kafka/src/main/java/com/heibaiying/springboot/bean/Programmer.java Normal file
View File

@ -0,0 +1,24 @@
package com.heibaiying.springboot.bean;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.io.Serializable;
import java.util.Date;
@Data
@AllArgsConstructor
@NoArgsConstructor
// 需要实现序列化接口
public class Programmer implements Serializable {
private String name;
private int age;
private float salary;
private Date birthday;
}

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

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

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

@ -0,0 +1,24 @@
package com.heibaiying.springboot.consumer;
import com.alibaba.fastjson.JSON;
import com.heibaiying.springboot.bean.Programmer;
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.stereotype.Component;
/**
* @author : heibaiying
* @description : kafka 消费者
*/
@Component
@Slf4j
public class KafkaBeanConsumer {
@KafkaListener(groupId = "group", topics = Topic.BEAN)
public void consumer(ConsumerRecord<String, Object> record) {
System.out.println("消费者收到消息:" + JSON.parseObject(record.value().toString(), Programmer.class));
}
}

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

@ -0,0 +1,37 @@
package com.heibaiying.springboot.consumer;
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.stereotype.Component;
/**
* @author : heibaiying
* @description : kafka 消费者组
*
* 多个消费者群组可以共同读取同一个主题，彼此之间互不影响。
* 但主题的一个分区只能被同一个消费者群组里面的一个消费者读取。
*/
@Component
@Slf4j
public class KafkaGroupConsumer {
// 分组1 中的消费者1
@KafkaListener(id = "consumer1-1", groupId = "group1", topics = Topic.GROUP)
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)
public void consumer1_2(ConsumerRecord<String, Object> record) {
System.out.println("consumer1-2 收到消息:" + 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());
}
}

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

@ -0,0 +1,25 @@
package com.heibaiying.springboot.consumer;
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.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
/**
* @author : heibaiying
* @description : kafka 简单消息消费者
*/
@Component
@Slf4j
public class KafkaSimpleConsumer {
// 简单消费者
@KafkaListener(groupId = "group",topics = Topic.SIMPLE)
public void consumer1_1(ConsumerRecord<String, Object> record, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
System.out.println("消费者收到消息:" + record.value() + "; topic:" + topic);
}
}

36
spring-boot/spring-boot-kafka/src/main/resources/application.yml Normal file
View File

@ -0,0 +1,36 @@
spring:
kafka:
# <20>Զ<EFBFBD><D4B6>ŷָ<C5B7><D6B8>ĵ<EFBFBD>ַ<EFBFBD>б<EFBFBD><D0B1><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ڽ<EFBFBD><DABD><EFBFBD><EFBFBD><EFBFBD>Kafka<6B><61>Ⱥ<EFBFBD>ij<EFBFBD>ʼ<EFBFBD><CABC><EFBFBD><EFBFBD>(kafka Ĭ<>ϵĶ˿ں<CBBF>Ϊ9092)
bootstrap-servers: 127.0.0.1:9092
producer:
# <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ϣ<EFBFBD>ط<EFBFBD><D8B7>Ĵ<EFBFBD><C4B4><EFBFBD><EFBFBD><EFBFBD>
retries: 0
#<23><><EFBFBD>ж<EFBFBD><D0B6><EFBFBD><EFBFBD><EFBFBD>Ϣ<EFBFBD><CFA2>Ҫ<EFBFBD><D2AA><EFBFBD><EFBFBD><EFBFBD>͵<EFBFBD>ͬһ<CDAC><D2BB><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ʱ<EFBFBD><CAB1><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>߻<EFBFBD><DFBB><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ƿ<EFBFBD><C7B7><EFBFBD>ͬһ<CDAC><D2BB><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ò<EFBFBD><C3B2><EFBFBD>ָ<EFBFBD><D6B8><EFBFBD><EFBFBD>һ<EFBFBD><D2BB><EFBFBD><EFBFBD><EFBFBD>ο<EFBFBD><CEBF><EFBFBD>ʹ<EFBFBD>õ<EFBFBD><C3B5>ڴ<EFBFBD><DAB4><EFBFBD>С<EFBFBD><D0A1><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ֽ<EFBFBD><D6BD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>㡣
batch-size: 16384
# <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ڴ滺<DAB4><E6BBBA><EFBFBD><EFBFBD><EFBFBD>Ĵ<EFBFBD>С<EFBFBD><D0A1>
buffer-memory: 33554432
# <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>л<EFBFBD><D0BB><EFBFBD>ʽ
key-serializer: org.apache.kafka.common.serialization.StringSerializer
# ֵ<><D6B5><EFBFBD><EFBFBD><EFBFBD>л<EFBFBD><D0BB><EFBFBD>ʽ
value-serializer: org.apache.kafka.common.serialization.StringSerializer
# acks=0 <20><> <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ڳɹ<DAB3>д<EFBFBD><D0B4><EFBFBD><EFBFBD>Ϣ֮ǰ<D6AE><C7B0><EFBFBD><EFBFBD><EFBFBD>ȴ<EFBFBD><C8B4>κ<EFBFBD><CEBA><EFBFBD><EFBFBD>Է<EFBFBD><D4B7><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ӧ<EFBFBD><D3A6>
# acks=1 <20><> ֻҪ<D6BB><D2AA>Ⱥ<EFBFBD><C8BA><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ڵ<EFBFBD><DAB5>յ<EFBFBD><D5B5><EFBFBD>Ϣ<EFBFBD><CFA2><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>߾ͻ<DFBE><CDBB>յ<EFBFBD>һ<EFBFBD><D2BB><EFBFBD><EFBFBD><EFBFBD>Է<EFBFBD><D4B7><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ɹ<EFBFBD><C9B9><EFBFBD>Ӧ<EFBFBD><D3A6>
# acks=all <20><>ֻ<EFBFBD>е<EFBFBD><D0B5><EFBFBD><EFBFBD>в<EFBFBD><D0B2>븴<EFBFBD>ƵĽڵ<C4BD>ȫ<EFBFBD><C8AB><EFBFBD>յ<EFBFBD><D5B5><EFBFBD>Ϣʱ<CFA2><CAB1><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>߲Ż<DFB2><C5BB>յ<EFBFBD>һ<EFBFBD><D2BB><EFBFBD><EFBFBD><EFBFBD>Է<EFBFBD><D4B7><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ijɹ<C4B3><C9B9><EFBFBD>Ӧ<EFBFBD><D3A6>
acks: all
consumer:
# <20>Զ<EFBFBD><D4B6>ύ<EFBFBD><E1BDBB>ʱ<EFBFBD><CAB1><EFBFBD><EFBFBD><EFBFBD><EFBFBD> <20><>spring boot 2.X <20>汾<EFBFBD><E6B1BE><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>õ<EFBFBD><C3B5><EFBFBD>ֵ<EFBFBD><D6B5><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ΪDuration <20><>Ҫ<EFBFBD><D2AA><EFBFBD><EFBFBD><EFBFBD>ض<EFBFBD><D8B6>ĸ<EFBFBD>ʽ<EFBFBD><CABD><EFBFBD><EFBFBD>1S,1M,2H,5D
auto-commit-interval: 1S
# <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ָ<EFBFBD><D6B8><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ڶ<EFBFBD>ȡһ<C8A1><D2BB>û<EFBFBD><C3BB>ƫ<EFBFBD><C6AB><EFBFBD><EFBFBD><EFBFBD>ķ<EFBFBD><C4B7><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ƫ<EFBFBD><C6AB><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ч<EFBFBD><D0A7><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>¸<EFBFBD><C2B8><EFBFBD><EFBFBD>δ<EFBFBD><CEB4><EFBFBD><EFBFBD><EFBFBD>
# latest<73><74>Ĭ<EFBFBD><C4AC>ֵ<EFBFBD><D6B5><EFBFBD><EFBFBD>ƫ<EFBFBD><C6AB><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ч<EFBFBD><D0A7><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>£<EFBFBD><C2A3><EFBFBD><EFBFBD><EFBFBD><EFBFBD>߽<EFBFBD><DFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>µļ<C2B5>¼<EFBFBD><C2BC>ʼ<EFBFBD><CABC>ȡ<EFBFBD><C8A1><EFBFBD>ݣ<EFBFBD><DDA3><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>֮<EFBFBD><D6AE><EFBFBD><EFBFBD><EFBFBD>ɵļ<C9B5>¼<EFBFBD><C2BC>
# earliest <20><><EFBFBD><EFBFBD>ƫ<EFBFBD><C6AB><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ч<EFBFBD><D0A7><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>£<EFBFBD><C2A3><EFBFBD><EFBFBD><EFBFBD><EFBFBD>߽<EFBFBD><DFBD><EFBFBD><EFBFBD><EFBFBD>ʼλ<CABC>ö<EFBFBD>ȡ<EFBFBD><C8A1><EFBFBD><EFBFBD><EFBFBD>ļ<EFBFBD>¼
auto-offset-reset: earliest
# <20>Ƿ<EFBFBD><C7B7>Զ<EFBFBD><D4B6>ύƫ<E1BDBB><C6AB><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ĭ<EFBFBD><C4AC>ֵ<EFBFBD><D6B5>true,Ϊ<>˱<EFBFBD><CBB1><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ظ<EFBFBD><D8B8><EFBFBD><EFBFBD>ݺ<EFBFBD><DDBA><EFBFBD><EFBFBD>ݶ<EFBFBD>ʧ<EFBFBD><CAA7><EFBFBD><EFBFBD><EFBFBD>԰<EFBFBD><D4B0><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ϊfalse,Ȼ<><C8BB><EFBFBD>ֶ<EFBFBD><D6B6>ύƫ<E1BDBB><C6AB><EFBFBD><EFBFBD>
enable-auto-commit: true
# <20><><EFBFBD>ķ<EFBFBD><C4B7><EFBFBD><EFBFBD>л<EFBFBD><D0BB><EFBFBD>ʽ
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
# ֵ<>ķ<EFBFBD><C4B7><EFBFBD><EFBFBD>л<EFBFBD><D0BB><EFBFBD>ʽ
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
listener:
# <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>е<EFBFBD><D0B5>߳<EFBFBD><DFB3><EFBFBD><EFBFBD><EFBFBD>
concurrency: 5

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

@ -0,0 +1,51 @@
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);
}
}
}