Spring Cloud Stream Kafka 基本使用
RocketMQ對(duì)Spring Cloud Stream 的介紹
主要概念
- 應(yīng)?用模型
- Binder 抽象
- 持久化 發(fā)布/訂閱?支持
- 消費(fèi)分組?支持
- 分區(qū)?支持
基本概念
Source：Stream 發(fā)送源
Sink：Stream 接收器器
Processor:
相關(guān)注解
激活:
- @EnableBinding
- @Con?guration
- @EnableIntegration
Source:
- @Output
- MessageChannel
Sink:
- @Input
- SubscribableChannel
- @ServiceActivator
- @StreamListener
生產(chǎn)者
配置
spring:
application:
name: stream-sink
cloud:
stream:
kafka:
binder:
brokers: localhost:9092
bindings:
goods-out: # 輸出通道
destination: goods # 對(duì)應(yīng)的topic
contentType: application/json
#也可以bing多個(gè)通道
log-out:
destination: log # 對(duì)應(yīng)的topic
contentType: application/json
default-binder: kafka #與consul 使用時(shí)需要指定 binder
定義通道
public interface GreetingsStreams {
String OUTPUT = "goods-out"; // 與配置中一樣
@Output(OUTPUT)
MessageChannel outboundGreetings();
}
激活
@EnableBinding(GreetingsStreams.class)
public class StreamsConfig {
}
發(fā)送消息
@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());
}
}
消費(fèi)者
配置
spring:
application:
name: stream-sink
cloud:
stream:
kafka:
binder:
brokers: localhost:9092
bindings:
goods-in:
destination: goods
contentType: application/json
group: finance # 指定消費(fèi)者組
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 從哪個(gè)分區(qū)獲取的數(shù)據(jù)
*/
@StreamListener(GreetingsStreams.INPUT)
public void handleGreetings(@Payload Greetings greetings,@Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition) {
log.info("Received message: {},from partition : {}", greetings,partition);
}
}
消費(fèi)分區(qū)
kafka 的partition 是一個(gè)有序隊(duì)列红竭，指定key可以將相同key的數(shù)據(jù)發(fā)送到同一個(gè)partition跛璧，可以保證消息有序消費(fèi)
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 表達(dá)式
partition-count: 4 # partition 數(shù)量
default-binder: kafka
發(fā)送端
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);
}
手動(dòng)應(yīng)答
消費(fèi)者端：
可以設(shè)置 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();
}
}
}
測(cè)試不成功焰坪，控制臺(tái)輸出的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 一直為自動(dòng)提交
生產(chǎn)端 可以 將應(yīng)答配置為同步的
spring.cloud.stream.kafka.bindings.output.producer.sync=true
https://docs.spring.io/spring-cloud-stream/docs/current/reference/htmlsingle/
