spring-cloud-stream学习
spring-cloud-stream学习
- 1. 简单使用
- 安装
- 测试例子
- 2. 使用其他中间件
- 测试例子
- 1. Source发送数据
- 配置 `application.yml`
- 创建 Source 应用
- 2. 实现 Processor（处理数据）
- 添加依赖
- 配置 `application.yml`
- 创建 Processor 应用
- 3.实现 Sink（消费数据）
- 添加依赖
- 配置 `application.yml`
- 创建 Sink 应用
- 启动步骤
- 配置与依赖注入
- 总结
原文网址：spring-cloud-stream学习
1. 简单使用
安装
curl -o 'cashcard.zip' 'https://start.spring.io/starter.zip?type=gradle-project&language=java&dependencies=cloud-stream%2Ckafka&name=CashCard&groupId=example&artifactId=cashcard&description=CashCard+service+for+Family+Cash+Cards&packaging=jar&packageName=example.cashcard&javaVersion=17' && unzip -d 'cashcard' 'cashcard.zip'
[~] $ cd cashcard
[~/cashcard] $
Next, run the ./gradlew build command:
[~/cashcard] $ ./gradlew build
[~/cashcard] $ cd build/libs/
[~/cashcard/build/libs] $ java -jar cashcard-0.0.1-SNAPSHOT.jar
测试例子
build.gradle
plugins {id 'java'id 'org.springframework.boot' version '3.2.5'id 'io.spring.dependency-management' version '1.1.4'
}group = 'example'
version = '0.0.1-SNAPSHOT'java {sourceCompatibility = '17'
}repositories {mavenCentral()
}ext {set('springCloudVersion', "2023.0.1")
}dependencies {implementation 'org.springframework.cloud:spring-cloud-stream'implementation 'org.springframework.cloud:spring-cloud-stream-binder-kafka'implementation 'org.springframework.kafka:spring-kafka'testImplementation 'org.springframework.boot:spring-boot-starter-test'testImplementation 'org.springframework.cloud:spring-cloud-stream-test-binder'testImplementation 'org.springframework.kafka:spring-kafka-test'
}dependencyManagement {imports {mavenBom "org.springframework.cloud:spring-cloud-dependencies:${springCloudVersion}"}
}tasks.named('test') {useJUnitPlatform()
}
test CashCardApplicationTests.java
package example.cashcard.stream;import java.io.IOException;import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.mock.mockito.MockBean;
import org.springframework.cloud.stream.binder.test.OutputDestination;
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
import org.springframework.context.annotation.Import;import example.cashcard.service.DataSourceService;
import static org.mockito.BDDMockito.given;
import example.cashcard.domain.CashCard;
import example.cashcard.domain.Transaction;import static org.assertj.core.api.Assertions.assertThat;
import org.springframework.messaging.Message;
import com.fasterxml.jackson.databind.ObjectMapper;@SpringBootTest
// Add this @Import statement
@Import(TestChannelBinderConfiguration.class)
class CashCardApplicationTests {// Autowire a mock bean for the DataSourceService@MockBeanprivate DataSourceService dataSourceService;@Testvoid basicCashCardSupplier1(@Autowired OutputDestination outputDestination) throws IOException {// Configure the mocked DataSourceServiceTransaction testTransaction = new Transaction(1L, new CashCard(123L, "sarah1", 1.00));given(dataSourceService.getData()).willReturn(testTransaction);// invoke the outputDestination and make sure it returned somethingMessage<byte[]> result = outputDestination.receive(5000, "approvalRequest-out-0");assertThat(result).isNotNull();// Deserialize the transaction and inspect itObjectMapper objectMapper = new ObjectMapper();Transaction transaction = objectMapper.readValue(result.getPayload(), Transaction.class);assertThat(transaction.id()).isEqualTo(1L);assertThat(transaction.cashCard()).isEqualTo(testTransaction.cashCard());}
}
程序 CashCardStream.java
package example.cashcard.stream;import example.cashcard.domain.Transaction;
import example.cashcard.service.DataSourceService;
import java.util.function.Supplier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class CashCardStream {@Beanpublic Supplier<Transaction> approvalRequest(DataSourceService dataSource) {// add this function callreturn () -> {return dataSource.getData();};}// Add this bean@Beanpublic DataSourceService dataSourceService() {return new DataSourceService();}
}
DataSourceService.java
package example.cashcard.service;import example.cashcard.domain.CashCard;
import example.cashcard.domain.Transaction;
import java.util.Random;public class DataSourceService {public Transaction getData() {CashCard cashCard = new CashCard(new Random().nextLong(), // Random ID"sarah1",new Random().nextDouble(100.00) // Random Amount);return new Transaction(new Random().nextLong(), cashCard);}
}
Transaction.java (record java14特性，16稳定版，只能构造函数添加数据，默认加了eq toString hash 构造函数，类是不可变的，字段值在构造后不能修改。)
package example.cashcard.domain;public record Transaction(Long id, CashCard cashCard) {}
CashCard.java
package example.cashcard.domain;public record CashCard(Long id,String owner,Double amountRequestedForAuth
) {}
运行
[~/exercises] $ docker ps
CONTAINER ID IMAGE COMMAND CREATED STATUS PORTS NAMES
0034eb13e35b apache/kafka:latest "/__cacert_entrypoin…" 15 minutes ago Up 15 minutes 0.0.0.0:9092->9092/tcp, :::9092->9092/tcp kafka
[~/exercises] $ ./gradlew bootRun
...
20XX-XX-XXTXX:XX:XX.XXXZ INFO 2627 --- [CashCard] [ main] o.a.kafka.common.utils.AppInfoParser : Kafka version: 3.6.2
20XX-XX-XXTXX:XX:XX.XXXZ INFO 2627 --- [CashCard] [ main] o.a.kafka.common.utils.AppInfoParser : Kafka commitId: c4deed513057c94e
20XX-XX-XXTXX:XX:XX.XXXZ INFO 2627 --- [CashCard] [ main] o.a.kafka.common.utils.AppInfoParser : Kafka startTimeMs: 1715638156031
20XX-XX-XXTXX:XX:XX.XXXZ INFO 2627 --- [CashCard] [ad | producer-1] org.apache.kafka.clients.Metadata : [Producer clientId=producer-1] Cluster ID: Some(5L6g3nShT-eMCtK--X86sw)
20XX-XX-XXTXX:XX:XX.XXXZ INFO 2627 --- [CashCard] [ main] o.s.c.s.m.DirectWithAttributesChannel : Channel 'CashCard.approvalRequest-out-0' has 1 subscriber(s).
20XX-XX-XXTXX:XX:XX.XXXZ INFO 2627 --- [CashCard] [ main] o.s.i.e.SourcePollingChannelAdapter : started bean 'approvalRequest-out-0_spca'
20XX-XX-XXTXX:XX:XX.XXXZ INFO 2627 --- [CashCard] [ main] example.cashcard.CashCardApplication : Started CashCardApplication in 1.624 seconds (process running for 1.812)
<==========---<==========---> 80% EXECUTING [59s]
> :bootRun
这个是1s打印一次
[~/exercises] $ docker exec -it kafka /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic approvalRequest-out-0
...
{"id":-66561165137247368,"cashCard":{"id":6094323678418692169,"owner":"sarah1","amountRequestedForAuth":50.7668781314909}}
{"id":-5634650775918976902,"cashCard":{"id":3090043269501770643,"owner":"sarah1","amountRequestedForAuth":63.79583001467617}}
{"id":-14741824561737749,"cashCard":{"id":-4335146560811993412,"owner":"sarah1","amountRequestedForAuth":12.783311898916805}}
{"id":5558799879234294766,"cashCard":{"id":-4613724913650180843,"owner":"sarah1","amountRequestedForAuth":58.932051104126955}}
{"id":6868476944436589763,"cashCard":{"id":8526417364307544245,"owner":"sarah1","amountRequestedForAuth":46.38569473593444}}
...
停止程序执行5s打印一次输出
[~/exercises] $ ./gradlew bootRun --args="--spring.integration.poller.fixed-delay=5000"
2. 使用其他中间件
[~/exercises] $ docker ps --format 'table {{.Names}}\t{{.ID}}\t{{.Image}}\t{{.Status}}'
NAMES CONTAINER ID IMAGE STATUS
kafka fde1261e8eab apache/kafka:latest Up 8 minutes
rabbitmq 27bce655e84d rabbitmq:management Up 8 minutes
[~/exercises] $ docker stop kafka
程序等会儿自动停止
...
Generating Transaction: Transaction[id=2175805216802058711, cashCard=CashCard[id=7436539443625164261, owner=sarah1, amountRequestedForAuth=2.112304484447358]]
Generating Transaction: Transaction[id=7686026790369954447, cashCard=CashCard[id=-9056617961033348083, owner=sarah1, amountRequestedForAuth=41.2613014355566]]
20XX-XX-XXTXX:XX:XX.XXXZ INFO 18618 --- [CashCard] [ad | producer-1] org.apache.kafka.clients.NetworkClient : [Producer clientId=producer-1] Node 1 disconnected.
20XX-XX-XXTXX:XX:XX.XXXZ INFO 18618 --- [CashCard] [ad | producer-1] org.apache.kafka.clients.NetworkClient : [Producer clientId=producer-1] Node -1 disconnected.
20XX-XX-XXTXX:XX:XX.XXXZ INFO 18618 --- [CashCard] [ad | producer-1] org.apache.kafka.clients.NetworkClient : [Producer clientId=producer-1] Node 1 disconnected.
20
[~/exercises] $ docker ps --format 'table {{.Names}}\t{{.ID}}\t{{.Image}}\t{{.Status}}'
NAMES CONTAINER ID IMAGE STATUS
rabbitmq 27bce655e84d rabbitmq:management Up 10 minutes
替换为rabbit
dependencies {implementation 'org.springframework.cloud:spring-cloud-stream-binder-rabbit'implementation 'org.springframework.cloud:spring-cloud-stream'// implementation 'org.springframework.cloud:spring-cloud-stream-binder-kafka'// implementation 'org.springframework.kafka:spring-kafka'testImplementation 'org.springframework.boot:spring-boot-starter-test'testImplementation 'org.springframework.cloud:spring-cloud-stream-test-binder'// testImplementation 'org.springframework.kafka:spring-kafka-test'
}
重启，发现每秒10条消息
[~/exercises] $ ./gradlew bootRun --args="--spring.integration.poller.fixed-delay=100"...
Generating Transaction: Transaction[id=-2719852727681158000, cashCard=CashCard[id=-6079371198316955842, owner=sarah1, amountRequestedForAuth=36.18919281848337]]
Generating Transaction: Transaction[id=-5607247743473931121, cashCard=CashCard[id=6889019574148834540, owner=sarah1, amountRequestedForAuth=0.7966132708258855]]
Generating Transaction: Transaction[id=8789939425625482805, cashCard=CashCard[id=6270568951233307661, owner=sarah1, amountRequestedForAuth=33.58135053399174]]
<==========---> 80% EXECUTING [8s]
> :bootRun
创建个队列
[~/exercises] $ docker exec -it rabbitmq sh -c "rabbitmqadmin declare queue name=test-queue && rabbitmqadmin declare binding source=approvalRequest-out-0 destination=test-queue routing_key=#"
使用web
dependencies {...implementation 'org.springframework.boot:spring-boot-starter-web'...
}
@PostMapping(path = "/publish/txn")
public void publishTxn(@RequestBody Transaction transaction) {System.out.println("POST for Transaction: " + transaction);
}
@SpringBootTest(webEnvironment = WebEnvironment.RANDOM_PORT)
public class CashCardControllerTests {...@Autowiredprivate TestRestTemplate restTemplate;@Testvoid cashCardStreamBridge() throws IOException {Transaction transaction = new Transaction(1L, new CashCard(123L, "kumar2", 1.00));ResponseEntity<Transaction> response = this.restTemplate.postForEntity("http://localhost:" + port + "/publish/txn",transaction, Transaction.class);assertThat(response.getStatusCode()).isEqualTo(HttpStatus.OK);}...
}
[~/exercises] $ ./gradlew test
...
BUILD SUCCESSFUL in 10s
4 actionable tasks: 4 executed
运行
[~/exercises] $ ./gradlew bootRun
...
20XX-XX-XXTXX:XX:XX.XXXZ INFO 14453 --- [CashCard] [ main] o.s.b.w.embedded.tomcat.TomcatWebServer : Tomcat started on port 8080 (http) with context path ''
20XX-XX-XXTXX:XX:XX.XXXZ INFO 14453 --- [CashCard] [ main] example.cashcard.CashCardApplication : Started CashCardApplication in 2.651 seconds (process running for 2.934)
<==========---<==========---> 80% EXECUTING [59s]
> :bootRun
测试
[~/exercises] $ curl -d '{"id" : 100,"cashCard" : {"id" : 209,"owner" : "kumar2","amountRequestedForAuth" : 200.0}
}' -H "Content-Type: application/json" -X POST http://localhost:8080/publish/txn
CashCardStream.java添加
package example.cashcard.stream;import example.cashcard.service.DataSourceService;
import example.cashcard.domain.Transaction;
import java.util.function.Supplier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.cloud.stream.function.StreamBridge;@Configuration
public class CashCardStream {//addprivate final StreamBridge streamBridge;//addpublic CashCardStream(StreamBridge streamBridge) {this.streamBridge = streamBridge;}//addpublic void publishOnDemand(Transaction transaction) {this.streamBridge.send("approvalRequest-out-0", transaction);}@Beanpublic Supplier<Transaction> approvalRequest(DataSourceService dataSource) {return () -> {return dataSource.getData();};}@Beanpublic DataSourceService dataSourceService() {return new DataSourceService();}
}CashCardController.java
package example.cashcard.controller;import example.cashcard.domain.Transaction;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;
import example.cashcard.stream.CashCardStream;@RestController
public class CashCardController {@PostMapping(path = "/publish/txn")public void publishTxn(@RequestBody Transaction transaction) {//addthis.cashCardStream.publishOnDemand(transaction);System.out.println("POST for Transaction: " + transaction);}
//addprivate CashCardStream cashCardStream;public CashCardController(CashCardStream cashCardStream) {this.cashCardStream = cashCardStream;}
}
测试类改为
package example.cashcard.controller;import static org.assertj.core.api.Assertions.assertThat;import java.io.IOException;import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.context.SpringBootTest.WebEnvironment;
import org.springframework.boot.test.web.client.TestRestTemplate;
import org.springframework.boot.test.web.server.LocalServerPort;
import org.springframework.cloud.stream.binder.test.OutputDestination;
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
import org.springframework.context.annotation.Import;
import org.springframework.messaging.Message;import com.fasterxml.jackson.databind.ObjectMapper;import example.cashcard.domain.CashCard;
import example.cashcard.domain.Transaction;
import example.cashcard.stream.CashCardStream;
//add
@SpringBootTest(webEnvironment = WebEnvironment.RANDOM_PORT)
@Import({ TestChannelBinderConfiguration.class, CashCardStream.class })
public class CashCardControllerTests {@LocalServerPortprivate int port;@Autowiredprivate TestRestTemplate restTemplate;//add@Testvoid postShouldSendTransactionAsAMessage(@Autowired OutputDestination outputDestination) throws IOException {Transaction postedTransaction = new Transaction(123L, new CashCard(1L, "Foo Bar", 1.00));this.restTemplate.postForEntity("http://localhost:" + port + "/publish/txn", postedTransaction, Transaction.class);
//addMessage<byte[]> result = outputDestination.receive(5000, "approvalRequest-out-0");assertThat(result).isNotNull();ObjectMapper objectMapper = new ObjectMapper();Transaction transactionFromMessage = objectMapper.readValue(result.getPayload(), Transaction.class);assertThat(transactionFromMessage.id()).isEqualTo(postedTransaction.id());}@SpringBootApplicationpublic static class App {}
}
测试失败了
[~/exercises] $ ./gradlew test --info
... lots and lots of output ...
CashCardControllerTests > cashCardStreamBridge(OutputDestination) FAILEDorg.opentest4j.AssertionFailedError:expected: 123Lbut was: -69970613332339299L
因为DataSourceService.java的getData产生了随机数，而着个都发送到一个交换机了，读取的时候读到了随机的那个。
重构下,创建CashCardTransactionOnDemand.java
package example.cashcard.ondemand;import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.context.annotation.Configuration;import example.cashcard.domain.Transaction;@Configuration
public class CashCardTransactionOnDemand {private final StreamBridge streamBridge;public CashCardTransactionOnDemand(StreamBridge streamBridge) {this.streamBridge = streamBridge;}public void publishOnDemand(Transaction transaction) {this.streamBridge.send("approvalRequest-out-0", transaction);}
}
改写CashCardStream.java
package example.cashcard.stream;import example.cashcard.service.DataSourceService;
import example.cashcard.domain.Transaction;
import java.util.function.Supplier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class CashCardStream {@Beanpublic Supplier<Transaction> approvalRequest(DataSourceService dataSource) {return () -> {return dataSource.getData();};}@Beanpublic DataSourceService dataSourceService() {return new DataSourceService();}
}
修改CashCardController.java
package example.cashcard.controller;
...
import example.cashcard.ondemand.CashCardTransactionOnDemand;
...@RestController
public class CashCardController {private final CashCardTransactionOnDemand cashCardTransactionOnDemand;public CashCardController(@Autowired CashCardTransactionOnDemand cashCardTransactionOnDemand) {this.cashCardTransactionOnDemand = cashCardTransactionOnDemand;}@PostMapping(path = "/publish/txn")public void publishTxn(@RequestBody Transaction transaction) {this.cashCardTransactionOnDemand.publishOnDemand(transaction);}
}
修改CashCardControllerTests.java
// update the two references from CashCardTransactionStream to CashCardTransactionOnDemand
...
import example.cashcard.ondemand.CashCardTransactionOnDemand;
...@SpringBootTest(webEnvironment = WebEnvironment.RANDOM_PORT)
@Import({TestChannelBinderConfiguration.class, CashCardTransactionOnDemand.class})class CashCardControllerTests {...
}
测试
[~/exercises] $ ./gradlew test
重写运行
./gradlew bootRun --args="--spring.integration.poller.fixed-delay=5000"
使用kafka验证
[~/exercises] $ docker exec -it kafka /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic approvalRequest-out-0
...
{"id":-66561165137247368,"cashCard":{"id":6094323678418692169,"owner":"sarah1","amountRequestedForAuth":50.7668781314909}}
{"id":-5634650775918976902,"cashCard":{"id":3090043269501770643,"owner":"sarah1","amountRequestedForAuth":63.79583001467617}}
...
使用post测试
[~/exercises] $ curl -d '{"id" : 100,"cashCard" : {"id" : 209,"owner" : "kumar2","amountRequestedForAuth" : 200.0}
}' -H "Content-Type: application/json" -X POST http://localhost:8080/publish/txn
你会看到post事务和自动生成的事务一起被立即推送到Kafka中。
...
{"id":-5884276558030210833,"cashCard":{"id":-2255250790195745652,"owner":"sarah1","amountRequestedForAuth":90.69339964569193}}
{"id":5266952642528070044,"cashCard":{"id":-5780440342191504864,"owner":"sarah1","amountRequestedForAuth":21.80623845666293}}
{"id":100,"cashCard":{"id":209,"owner":"kumar2","amountRequestedForAuth":200.0}}
{"id":100,"cashCard":{"id":209,"owner":"kumar2","amountRequestedForAuth":200.0}}
{"id":100,"cashCard":{"id":209,"owner":"kumar2","amountRequestedForAuth":200.0}}
{"id":-4042360335249930640,"cashCard":{"id":-8391359726931016778,"owner":"sarah1","amountRequestedForAuth":85.5669171882778}}
测试例子
1. Source发送数据
<dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>
配置 application.yml
spring:application:name: source-appcloud:stream:bindings:output:destination: raw-datacontent-type: application/json
server:port: 8081
创建 Source 应用
SourceApplication.java
@SpringBootApplication
public class SourceApplication {public static void main(String[] args) {SpringApplication.run(SourceApplication.class, args);}
}
MessageSource.java
@Component
public class MessageSource {@StreamEmitterpublic void emit() {Map<String, Object> message = new HashMap<>();message.put("orderId", "12345");message.put("amount", 100);// 发送消息到 raw-data 队列Source.output().send(MessageBuilder.withPayload(message).build());}
}
2. 实现 Processor（处理数据）
添加依赖
<dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>
配置 application.yml
spring:application:name: processor-appcloud:stream:bindings:input:destination: raw-datacontent-type: application/jsonoutput:destination: enriched-datacontent-type: application/json
server:port: 8082
创建 Processor 应用
ProcessorApplication.java
@SpringBootApplication
public class ProcessorApplication {public static void main(String[] args) {SpringApplication.run(ProcessorApplication.class, args);}
}
DataProcessor.java
@Component
public class DataProcessor {@Beanpublic Function<Map<String, Object>, Map<String, Object>> enrich() {return input -> {input.put("processedTime", Instant.now().toString());input.put("enriched", true);return input;};}
}
3.实现 Sink（消费数据）
添加依赖
<dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>
配置 application.yml
spring:application:name: sink-appcloud:stream:bindings:input:destination: enriched-datacontent-type: application/json
server:port: 8083
创建 Sink 应用
SinkApplication.java
@SpringBootApplication
public class SinkApplication {public static void main(String[] args) {SpringApplication.run(SinkApplication.class, args);}
}
DataSink.java
@Component
public class DataSink {@Beanpublic Consumer<Map<String, Object>> logSink() {return data -> {System.out.println("接收到的数据已落地处理：" + data);};}
}
启动步骤
- 启动 Source 应用： 运行
SourceApplication，它会定时向raw-data队列发送消息。 - 启动 Processor 应用： 运行
ProcessorApplication，它会从raw-data队列接收消息，进行处理，并将增强后的数据发送到enriched-data队列。 - 启动 Sink 应用： 运行
SinkApplication，它会从enriched-data队列接收消息，并打印或处理这些消息。
配置与依赖注入
- Source 组件：负责生成消息并发送到 RabbitMQ 队列。
- Processor 组件：接收消息并进行数据处理（如添加字段、数据转换等）。
- Sink 组件：消费处理后的数据并落地（如打印、数据库存储等）。
每个应用都会使用 RabbitMQ 队列作为通信桥梁，raw-data 队列连接 Source 和 Processor，enriched-data 队列连接 Processor 和 Sink。
总结
- 你已经成功将 Source、Processor 和 Sink 拆分成三个独立的 Spring Boot 应用，它们通过 RabbitMQ 队列进行通信。
- 每个应用专注于单一功能（生产、处理、消费），符合微服务架构的设计原则。
- 你可以根据业务需求扩展各个服务并独立部署、测试和维护。