- \n
- \n
- \u958b\u767a\u74b0\u5883: Mac<\/ul>\n<\/li>\n<\/ul>\n
- \n
- IDE: IntelliJ<\/ul>\n<\/li>\n<\/ul>\n
- \n
- Spring Boot: 2.0.3<\/ul>\n<\/li>\n<\/ul>\n
- \n
- Web -> Reactive Web<\/ul>\n<\/li>\n<\/ul>\n
<\/p>\n
- Cloud Messaging -> Reactive Cloud Stream<\/ul>\n
![\"image.png\"](\"https:\/\/cdn.silicloud.com\/blog-img\/blog\/img\/657d7b08913a08637a697e63\/18-0.png\" "\\"\\"")
<\/div>\n\u9879\u76ee\u540d\u79f0\uff0c\u5730\u70b9<\/h3>\n
\u8bf7\u968f\u610f\u586b\u5199\u59d3\u540d\u548c\u4f4d\u7f6e\u4ee5\u5b8c\u6210\u3002<\/p>\n
![\"image.png\"](\"https:\/\/cdn.silicloud.com\/blog-img\/blog\/img\/657d7b08913a08637a697e63\/21-0.png\" "\\"\\"")
<\/div>\n\u5728\u9879\u76ee\u6b63\u5728\u521b\u5efa\u4e2d\uff0c\u8bf7\u7a0d\u7b49\u4e00\u4e0b\u3002<\/p>\n
\u5f53\u5b8c\u6210\u65f6\uff0c\u4f1a\u53d8\u6210\u8fd9\u79cd\u611f\u89c9\u3002<\/p>\n
![\"image.png\"](\"https:\/\/cdn.silicloud.com\/blog-img\/blog\/img\/657d7b08913a08637a697e63\/24-0.png\" "\\"\\"")
<\/div>\n\u68c0\u67e5\u548c\u4fee\u6b63Pom\u6587\u4ef6<\/h2>\n
\u8bf7\u786e\u8ba4\u901a\u8fc7SPRING INITIALIZR\u521b\u5efa\u7684\u9879\u76ee\u7684pom\u6587\u4ef6\u7684\u8bbe\u7f6e\u5185\u5bb9\u3002<\/h3>\n
spring-cloud.version\u5e94\u8be5\u6307\u5b9a\u4e3aFinchley.RELEASE<\/p>\n
<properties>\r\n ...\r\n <spring-cloud.version>Finchley.RELEASE<\/spring-cloud.version>\r\n <\/properties>\r\n<\/code><\/pre>\n\u5e94\u8be5\u6307\u5b9a\u4e86\u4e2d\u7684spring-boot-starter-webflux\u548cspring-cloud-stream-reactive\u3002<\/p>\n
<dependencies>\r\n <dependency>\r\n <groupId>org.springframework.boot<\/groupId>\r\n <artifactId>spring-boot-starter-webflux<\/artifactId>\r\n <\/dependency>\r\n ...\r\n <dependency>\r\n <groupId>org.springframework.cloud<\/groupId>\r\n <artifactId>spring-cloud-stream-reactive<\/artifactId>\r\n <\/dependency>\r\n ...\r\n <\/dependencies>\r\n<\/code><\/pre>\n\u6dfb\u52a0 Kafka \u8fde\u63a5\u5668\u7684\u914d\u7f6e\u81f3 pom \u6587\u4ef6\u3002<\/h3>\n
\u8bf7\u6dfb\u52a0 spring-cloud-starter-stream-kafka\u3002
\n\u6ce8\u610f\uff0c\u5982\u679c\u6ca1\u6709\u8fd9\u4e2a\u914d\u7f6e\uff0c\u5c06\u65e0\u6cd5\u8fde\u63a5\u5230Kafka\u3002<\/p>\n<dependency>\r\n <groupId>org.springframework.cloud<\/groupId>\r\n <artifactId>spring-cloud-starter-stream-kafka<\/artifactId>\r\n <\/dependency>\r\n\r\n<\/code><\/pre>\n\u6dfb\u52a0\u5143\u7d20\uff0c\u4ee5\u4fbf\u80fd\u591f\u52a0\u8f7dspring-cloud-starter-stream-kafka\u3002<\/p>\n
<repositories>\r\n <repository>\r\n <id>spring-milestones<\/id>\r\n <name>Spring Milestones<\/name>\r\n <url>https:\/\/repo.spring.io\/milestone<\/url>\r\n <snapshots>\r\n <enabled>false<\/enabled>\r\n <\/snapshots>\r\n <\/repository>\r\n <\/repositories>\r\n<\/code><\/pre>\n\u6dfb\u52a0\u4ee3\u7801<\/h2>\n
\u6587\u4ef6\u7ed3\u6784<\/h3>\n
\u6dfb\u52a0 Routes.kt\u3001Handler.kt \u548c Producer.kt\u3002<\/p>\n
21:03:59 ~\/IdeaProjects\/spring-cloud-stream-kafka-demo\/src\/main\/kotlin\/com\/example\/springcloudstreamkafkademo \r\n$ tree\r\n.\r\n\u251c\u2500\u2500 Handler.kt\r\n\u251c\u2500\u2500 Producer.kt\r\n\u251c\u2500\u2500 Routes.kt\r\n\u2514\u2500\u2500 SpringCloudStreamKafkaDemoApplication.kt\r\n<\/code><\/pre>\nRoutes.kt \u7684\u4e2d\u6587\u8bd1\u6587\u53ef\u4ee5\u662f\uff1a\u8def\u7531.kt<\/h3>\n
\u5728\/ api \/ putText\u7684\u8def\u5f84\u4e0a\uff0c\u63a5\u53d7JSON\u8bf7\u6c42\u5e76\u8c03\u7528\u5904\u7406\u7a0b\u5e8f(Handler)\u3002<\/p>\n
package<\/span> com.example.springcloudstreamkafkademo<\/span>\r\n\r\nimport<\/span> org.springframework.context.annotation.Bean<\/span>\r\nimport<\/span> org.springframework.context.annotation.Configuration<\/span>\r\nimport<\/span> org.springframework.http.MediaType<\/span>\r\n\r\n@Configuration<\/span>\r\nclass<\/span> Routes<\/span>(<\/span>\r\n private<\/span> val<\/span> handler<\/span>:<\/span> Handler<\/span>\r\n)<\/span> {<\/span>\r\n\r\n @Bean<\/span>\r\n fun<\/span> router<\/span>()<\/span> =<\/span> org<\/span>.<\/span>springframework<\/span>.<\/span>web<\/span>.<\/span>reactive<\/span>.<\/span>function<\/span>.<\/span>server<\/span>.<\/span>router<\/span> {<\/span>\r\n \"\/api\"<\/span>.<\/span>nest<\/span> {<\/span>\r\n accept<\/span>(<\/span>MediaType<\/span>.<\/span>APPLICATION_JSON<\/span>).<\/span>nest<\/span> {<\/span>\r\n POST<\/span>(<\/span>\"\/putText\"<\/span>,<\/span> handler<\/span>::<\/span>output<\/span>)<\/span>\r\n }<\/span>\r\n }<\/span>\r\n }<\/span>\r\n}<\/span>\r\n<\/code><\/pre>\n\u5904\u7406\u7a0b\u5e8f.kt<\/h3>\n
\u5c06\u8bf7\u6c42\u7684JSON\u8f6c\u6362\u4e3a\u5bf9\u8c61\uff08RequestMessage\uff09\uff0c\u5e76\u4f20\u9012\u7ed9Kafka\u751f\u4ea7\u8005\u3002<\/p>\n
package<\/span> com.example.springcloudstreamkafkademo<\/span>\r\n\r\nimport<\/span> org.springframework.stereotype.Component<\/span>\r\nimport<\/span> org.springframework.web.reactive.function.server.ServerRequest<\/span>\r\nimport<\/span> org.springframework.web.reactive.function.server.ServerResponse<\/span>\r\nimport<\/span> org.springframework.web.reactive.function.server.body<\/span>\r\nimport<\/span> reactor.core.publisher.Mono<\/span>\r\n\r\ndata class<\/span> RequestMessage<\/span>(<\/span>\r\n val<\/span> id<\/span>:<\/span> Long<\/span>,<\/span>\r\n val<\/span> text<\/span>:<\/span> String<\/span>\r\n)<\/span>\r\n\r\n@Component<\/span>\r\nclass<\/span> Handler<\/span>(<\/span>private<\/span> val<\/span> producer<\/span>:<\/span> Producer<\/span>)<\/span> {<\/span>\r\n\r\n fun<\/span> output<\/span>(<\/span>req<\/span>:<\/span> ServerRequest<\/span>)<\/span> =<\/span>\r\n req<\/span>.<\/span>bodyToMono<\/span>(<\/span>RequestMessage<\/span>::<\/span>class<\/span>.<\/span>java<\/span>)<\/span>\r\n .<\/span>flatMap<\/span> {<\/span> ServerResponse<\/span>.<\/span>ok<\/span>().<\/span>body<\/span>(<\/span>Mono<\/span>.<\/span>just<\/span>(<\/span>producer<\/span>.<\/span>output<\/span>(<\/span>it<\/span>)))<\/span> }<\/span>\r\n\r\n}<\/span>\r\n<\/code><\/pre>\n\u5236\u7247\u4eba.kt<\/h3>\n
\u8fd9\u91cc\u662f\u5173\u952e\u3002\u5bf9\u4e8e\u8bf4\u660e\uff0c\u53ef\u4ee5\u53c2\u8003Spring Cloud Stream\u53c2\u8003\u6307\u5357\u3002\u7136\u800c\uff0c\u5728\u7ecf\u8fc7\u591a\u6b21\u5c1d\u8bd5\u4e4b\u540e\uff0c\u6211\u627e\u5230\u4e86\u4e00\u79cd\u6709\u6548\u7684\u65b9\u6cd5\uff0c\u4f46\u4e5f\u53ef\u80fd\u6709\u5176\u4ed6\u66f4\u597d\u7684\u65b9\u5f0f\u3002<\/p>\n
\u5728output\u51fd\u6570\u4e2d\uff0c\u6211\u4eec\u5c06\u6d88\u606f\u4f20\u9012\u7ed9fluxSink\u3002\u800cfluxSink\u662f\u5728\u751f\u6210Flux\u65f6\u83b7\u53d6\u7684emitter\uff0c\u7531flux\u7684emit\u51fd\u6570\u8ba2\u9605\u540e\uff0c\u4f1a\u6309\u987a\u5e8f\u5904\u7406fluxSink\u4f20\u9012\u7684\u6d88\u606f\u3002<\/p>\n
package<\/span> com.example.springcloudstreamkafkademo<\/span>\r\n\r\nimport<\/span> org.springframework.cloud.stream.annotation.EnableBinding<\/span>\r\nimport<\/span> org.springframework.cloud.stream.annotation.Output<\/span>\r\nimport<\/span> org.springframework.cloud.stream.messaging.Source<\/span>\r\nimport<\/span> org.springframework.cloud.stream.reactive.StreamEmitter<\/span>\r\nimport<\/span> org.springframework.stereotype.Component<\/span>\r\nimport<\/span> reactor.core.publisher.Flux<\/span>\r\nimport<\/span> reactor.core.publisher.FluxSink<\/span>\r\n\r\n@Component<\/span>\r\n@EnableBinding<\/span>(<\/span>Source<\/span>::<\/span>class<\/span>)<\/span>\r\nclass<\/span> Producer<\/span> {<\/span>\r\n\r\n private<\/span> var<\/span> fluxSink<\/span>:<\/span> FluxSink<\/span><<\/span>RequestMessage<\/span>>?<\/span> =<\/span> null<\/span>\r\n\r\n private<\/span> var<\/span> flux<\/span>:<\/span> Flux<\/span><<\/span>RequestMessage<\/span>><\/span> =<\/span> Flux<\/span>.<\/span>create<\/span><<\/span>RequestMessage<\/span>>(<\/span>\r\n {<\/span> emitter<\/span> -><\/span> this<\/span>.<\/span>fluxSink<\/span> =<\/span> emitter<\/span> },<\/span>\r\n FluxSink<\/span>.<\/span>OverflowStrategy<\/span>.<\/span>BUFFER<\/span>)<\/span>\r\n\r\n fun<\/span> output<\/span>(<\/span>message<\/span>:<\/span> RequestMessage<\/span>):<\/span> RequestMessage<\/span> {<\/span>\r\n this<\/span>.<\/span>
<\/p>\n
- \n
- \n
<\/p>\n
- \u8a00\u8a9e: Kotlin<\/ul>\n
\u9884\u5148\u51c6\u5907<\/h1>\n
\u5728Mac\u4e0a\u5b89\u88c5Kafka\u3002<\/h2>\n
\u8bf7\u6309\u7167\u4ee5\u4e0b\u975e\u5e38\u7b80\u660e\u7684\u6b65\u9aa4\u5728Mac\u4e0a\u5b89\u88c5Kafka\uff08Kafka\u662f\u4e00\u79cd\u6d88\u606f\u961f\u5217\u8f6f\u4ef6\uff09\u3002<\/p>\n
\u64cd\u4f5c\u6b65\u9aa4<\/h1>\n
\u9879\u76ee\u521b\u5efa<\/h2>\n
\u6211\u5c06\u4f7f\u7528IntelliJ\uff0c\u4eceSPRING INITIALIZR\u4e2d\u521b\u5efa\u4e00\u4e2aSpring Boot\u9879\u76ee\u3002<\/p>\n
\u6587\u4ef6 -> \u65b0\u5efa -> \u9879\u76ee…<\/h3>\n
![\"image.png\"](\"https:\/\/cdn.silicloud.com\/blog-img\/blog\/img\/657d7b08913a08637a697e63\/12-0.png\" "\\"\\"")
<\/div>\n\u9879\u76ee\u5143\u6570\u636e<\/h3>\n
- Language: Kotlin<\/ul>\n
![\"image.png\"](\"https:\/\/cdn.silicloud.com\/blog-img\/blog\/img\/657d7b08913a08637a697e63\/15-0.png\" "\\"\\"")
<\/div>\n\u4f9d\u8d56\u5173\u7cfb<\/h3>\n
- \n
- \n
<\/p>\n
- \n
- \n