Spring Cloud Stream Kafka 基本使用
RocketMQ对Spring Cloud Stream 的介绍
主要概念
- 应⽤用模型
- Binder 抽象
- 持久化 发布/订阅⽀支持
- 消费分组⽀支持
- 分区⽀支持
基本概念
Source：Stream 发送源
Sink：Stream 接收器器
Processor:
相关注解
激活:
- @EnableBinding
- @Configuration
- @EnableIntegration
Source:
- @Output
- MessageChannel
Sink:
- @Input
- SubscribableChannel
- @ServiceActivator
- @StreamListener
生产者
配置
spring:
application:
name: stream-sink
cloud:
stream:
kafka:
binder:
brokers: localhost:9092
bindings:
goods-out: # 输出通道
destination: goods # 对应的topic
contentType: application/json
#也可以bing多个通道
log-out:
destination: log # 对应的topic
contentType: application/json
default-binder: kafka #与consul 使用时需要指定 binder
定义通道
public interface GreetingsStreams {
String OUTPUT = "goods-out"; // 与配置中一样
@Output(OUTPUT)
MessageChannel outboundGreetings();
}
激活
@EnableBinding(GreetingsStreams.class)
public class StreamsConfig {
}
发送消息
@Service
@Slf4j
public class GreetingsService {
private final GreetingsStreams greetingsStreams;
public GreetingsService(GreetingsStreams greetingsStreams) {
this.greetingsStreams = greetingsStreams;
}
public void sendGreeting(final Greetings greetings) {
log.info("Sending greetings {}", greetings);
MessageChannel messageChannel = greetingsStreams.outboundGreetings();
messageChannel.send(MessageBuilder
.withPayload(greetings)
.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
.build());
}
}
消费者
配置
spring:
application:
name: stream-sink
cloud:
stream:
kafka:
binder:
brokers: localhost:9092
bindings:
goods-in:
destination: goods
contentType: application/json
group: finance # 指定消费者组
default-binder: kafka
定义通道
public interface GreetingsStreams {
String INPUT = "goods-in";
@Input(INPUT)
SubscribableChannel inboundGreetings();
}
激活
@EnableBinding(GreetingsStreams.class)
public class StreamsConfig {
}
接收消息
@Component
@Slf4j
public class GreetingsListener {
/**
*
* @param greetings
* @param partition 从哪个分区获取的数据
*/
@StreamListener(GreetingsStreams.INPUT)
public void handleGreetings(@Payload Greetings greetings,@Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition) {
log.info("Received message: {},from partition : {}", greetings,partition);
}
}
消费分区
kafka 的partition 是一个有序队列，指定key可以将相同key的数据发送到同一个partition，可以保证消息有序消费
spring:
application:
name: stream-source
cloud:
stream:
kafka:
binder:
brokers: localhost:9092
auto-add-partitions: true
bindings:
goods-out:
destination: goods
contentType: application/json
producer:
partition-key-expression: headers['partitionKey'] # partition key 表达式
partition-count: 4 # partition 数量
default-binder: kafka
发送端
private final static String PARTITION_KEY = "partitionKey";
private final GreetingsStreams greetingsStreams;
public GreetingsService(GreetingsStreams greetingsStreams) {
this.greetingsStreams = greetingsStreams;
}
public void sendGreeting(final Greetings greetings, String key) {
log.info("Sending greetings {}", greetings);
MessageChannel messageChannel = greetingsStreams.outboundGreetings();
messageChannel.send(MessageBuilder
.withPayload(greetings)
.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
.setHeader(PARTITION_KEY, key)
.build());
}
接收端
@StreamListener( target = GreetingsStreams.INPUT, condition = "headers['partitionKey']=='2'")
可以指定condition，接收指定条件的消息
@StreamListener( target = GreetingsStreams.INPUT, condition = "headers['partitionKey']=='1'")
public void handleKey1(@Payload Greetings greetings,@Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition) {
log.info("Received message: {},from partition : {}", greetings,partition);
}
@StreamListener( target = GreetingsStreams.INPUT, condition = "headers['partitionKey']=='2'")
public void handleKey2(@Payload Greetings greetings,@Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition) {
log.info("Received message: {},from partition : {}", greetings,partition);
}
@StreamListener( target = GreetingsStreams.INPUT)
public void handle(@Payload Greetings greetings,@Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition) {
log.info("Received message: {},from partition : {}", greetings,partition);
}
手动应答
消费者端：
可以设置 spring.cloud.stream.kafka.bindings.input.consumer.autoCommitOffset 为false
@SpringBootApplication
@EnableBinding(Sink.class)
public class ManuallyAcknowdledgingConsumer {
public static void main(String[] args) {
SpringApplication.run(ManuallyAcknowdledgingConsumer.class, args);
}
@StreamListener(Sink.INPUT)
public void process(Message<?> message) {
Acknowledgment acknowledgment = message.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class);
if (acknowledgment != null) {
System.out.println("Acknowledgment provided");
acknowledgment.acknowledge();
}
}
}
测试不成功，控制台输出的consumerConfig 依然为true,不知道为啥
auto.commit.interval.ms = 5000
auto.offset.reset = latest
bootstrap.servers = [localhost:9092]
check.crcs = true
client.id =
connections.max.idle.ms = 540000
default.api.timeout.ms = 60000
enable.auto.commit = true 一直为自动提交
生产端 可以 将应答配置为同步的
spring.cloud.stream.kafka.bindings.output.producer.sync=true
https://docs.spring.io/spring-cloud-stream/docs/current/reference/htmlsingle/
