paulwong

2021年12月1日 #

SPRING CLOUD - REACTIVE FEIGN

目前SPRING CLOUD(2020)尚未支持REACTIVE FEIGN,但官方推荐使用feign-reactive。

pom.xml
<dependency>
    <groupId>com.playtika.reactivefeign</groupId>
    <artifactId>feign-reactor-spring-cloud-starter</artifactId>
    <version>3.1.2</version>
    <type>pom</type>
</dependency>


LoanDecisionClientReactive.java
package com.paul.testspringcloudstream.loancheck.service;

import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;

import com.paul.testspringcloudstream.common.model.Loan;

import reactivefeign.spring.config.ReactiveFeignClient;
import reactor.core.publisher.Mono;

@ReactiveFeignClient(name = "loan-decision")
public interface LoanDecisionClientReactive {
    
    @PostMapping("/loan-decision")
    public Mono<Loan> getDecision(@RequestBody Loan loan);

}

LoanCheckConfiguration.java
@Configuration
@Import({
    MongoDbConsumerConfiguration.class,
})
@EnableDiscoveryClient
@EnableReactiveFeignClients("com.paul.testspringcloudstream.loancheck.service")
public class LoanCheckConfiguration {
}


使用同feign
@Autowired
private LoanDecisionClientReactive loanDecisionClientReactive;

Reference
https://blog.csdn.net/LCBUSHIHAHA/article/details/113817966

官方SAMPLE
https://github.com/kptfh/feign-reactive-sample

posted @ 2021-12-01 16:45 paulwong 阅读(73) | 评论 (0)编辑 收藏

Spring bootstrap.yml 不管用,失效解决方案(spring cloud 2020.x.x)

升级spring cloud版本之后发现bootstrap.yml 失效了,阅读官方文档得知,需要新增一个引用来开启bootstrap.xml文件的读取,新版spring cloud默认是关闭读取了。
增加依赖如下即可:
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-bootstrap</artifactId>
</dependency>
官方文档:
https://docs.spring.io/spring-cloud-config/docs/current/reference/html/#config-first-bootstrap

posted @ 2021-12-01 10:29 paulwong 阅读(10) | 评论 (0)编辑 收藏

2021年11月23日 #

SPRING REACTOR 之Flux和Mono

SPRING REACTOR 之Flux和Mono,有点象SPRING INTEGRATION的IntegrationFlow,有如下特点
  • 定义了针对某种类型数据的处理流程
  • 可以进行类型转换
  • 长期运行,除非被要求中止
  • 流程中的每种操作可以在新的线程中执行
  • 可以正常中止,如果中途有异常,则该流程也会中止
  • 要subscribe,流程才开始被启动
  • 可以分割成各个子流程
  • 可以聚合子流程
  • Mono发送一个数据,就发送中止信号
  • Flux发送任意数据,由程序决定何时发送中止信号

编程则比较简单,先根据不同的数据类型定义不同的Flux或Mono,业务操作用Function包装后,放在map/flatmap中,再调用subscribe启动流程。

posted @ 2021-11-23 14:30 paulwong 阅读(97) | 评论 (0)编辑 收藏

SPRING REACTOR 使用样例

SpringReactorTest.java

package com.paul.testreactivestream.reactor;


import java.util.List;

import org.junit.jupiter.api.Test;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

public class SpringReactorTest {
    
    private void subscribeAndEnd(Flux<?> flux) {
        
        flux.map(c -> String.format("[%s] %s", Thread.currentThread().getName(), c))
            .subscribe(System.out::println);
        
        flux.blockLast();
    }
    
    @Test
    public void createAFlux_just() throws InterruptedException {
        Flux<String> fruitFlux = 
                Flux.just("Apple", "Orange", "Grape", "Banana", "Strawberry")
                    .log()
                    ;
        fruitFlux.subscribe(
                     f -> System.out.println(
                                 String.format("[%s] Here's some fruit: %s", Thread.currentThread().getName(), f)
                             )
                  )
                 ;
        fruitFlux.blockLast();
        
//        Thread.currentThread().join();
    }
    
    @Test
    public void zipFluxesToObject() {
        Flux<String> characterFlux = 
                Flux.just("Garfield", "Kojak", "Barbossa");
        
        Flux<String> foodFlux = 
                Flux.just("Lasagna", "Lollipops", "Apples");
        
        Flux<String> zippedFlux = 
                Flux.zip(characterFlux, foodFlux, (c, f) -> c + " eats " + f);
        
        this.subscribeAndEnd(zippedFlux);
    }
    
    @Test
    public void map() {
        Flux<Player> playerFlux = 
                Flux.just("Michael Jordan", "Scottie Pippen", "Steve Kerr")
                    .map(n -> {
                        String[] split = n.split("\\s");
                        return new Player(split[0], split[1]);
                    })
                    ;
        this.subscribeAndEnd(playerFlux);
    }
    
    @Test
    public void flatMap() {
        Flux<Player> playerFlux = 
                Flux.just("Michael Jordan", "Scottie Pippen", "Steve Kerr")
                    .flatMap(
                        n -> Mono.just(n)
                                 .map(p -> {
                                    String[] split = p.split("\\s");
                                    return new Player(split[0], split[1]);
                                  })
                                 .subscribeOn(Schedulers.parallel())
                     );
        this.subscribeAndEnd(playerFlux);
    }
    
    @Test
    public void buffer() {
        Flux<List<String>> fruitFlux = 
                Flux.just(
                        "apple", "orange", "banana", "kiwi", "strawberry"
                     )
                    .buffer(3);
        this.subscribeAndEnd(fruitFlux);
    }
    
    @Test
    public void bufferAsyn() {
        Flux<String> flux =
            Flux.just(
                    "apple", "orange", "banana", "kiwi", "strawberry"
                 )
                .buffer(3)
                .flatMap(x ->
                    Flux.fromIterable(x)
                        .map(y -> y.toUpperCase())
                        .subscribeOn(Schedulers.parallel())
    //                    .log()
                 );
        this.subscribeAndEnd(flux);
    }
    
    @Test
    public void all() {
        Mono<Boolean> animalFlux = 
                Flux.just(
                        "aardvark", "elephant", "koala", "eagle", "kangaroo"
                     )
                    .all(c -> c.contains("a"))
                    ;
        animalFlux.map(c -> String.format("[%s] %s", Thread.currentThread().getName(), c))
                  .subscribe(System.out::println);
    
    }

}

posted @ 2021-11-23 13:59 paulwong 阅读(83) | 评论 (0)编辑 收藏

SPRING REACTOR资源

Spring Reactor 入门与实践
https://www.jianshu.com/p/7ee89f70dfe5

Project Reactor 核心原理解析
https://www.jianshu.com/p/df395eb28f69

Project Reactor 之 publishOn 与 subscribeOn
https://www.jianshu.com/p/769f6e9824fb

Spring响应式编程
https://blog.csdn.net/get_set/category_7484996.html









posted @ 2021-11-23 13:51 paulwong 阅读(11) | 评论 (0)编辑 收藏

EVEN DRIVEN - SPRING CLOUD STREAM - @PollableBean for Reactive Suppliers

Supplier beans, or functions that only publish messages in Spring Cloud Stream, are a bit special in that they aren't triggered by the receiving of events like Function or Consumer beans. This means that you often need a way to trigger them to be executed periodically.

For imperative functions the framework by default "polls" a Supplier function every 1 second, but that duration is configurable using the spring.cloud.stream.poller.fixed-delay property.

However, for reactive functions supplying a Flux it is only triggered once by default. This is because a Flux itself is potentially an infinite stream of events so in many cases it will only need to be triggered once. But don't worry, if you want to periodically trigger a reactive Supplier because you are producing a finite stream of events you can still do so using @PollableBean. This annotation then allows you to configure how often the function is triggered using the same spring.cloud.stream.poller.fixed-delay property!

One example use case here could be periodically querying a data store and publishing each entry/row as an event. The number of rows in your data store is a finite number at any given time.

Example code:

@PollableBean 
public Supplier<Flux<String>> stringSupplier() { return () -> Flux.just("foo","bar","baz"); }


Reference:
https://solace.community/discussion/360/pollablebean-for-reactive-suppliers-in-spring-cloud-stream

posted @ 2021-11-23 10:03 paulwong 阅读(18) | 评论 (0)编辑 收藏

2021年11月19日 #

EVEN DRIVEN - SPRING CLOUD STREAM - 从非SCS组件发送消息到SCS组件

在SPRING INTEGRATION中,如果要从非SPRING INTEGRATION代码发送MESSAGE到SPRING INTEGRATION程序,通常用BUS GATEWAY。

那么在SPRING CLOUD STREAM中,如果要从非SPRING CLOUD STREAM代码发送MESSAGE到SPRING CLOUD STREAM程序,通常就要先通知框架自动生成一个SOURCE。

application.property
spring.cloud.stream.source=supplier
spring.cloud.stream.bindings.supplier-out-0.destination=notification-events

java
streamBridge.send("supplier-out-0", userDto);

Reference:
https://blog.devgenius.io/event-driven-microservices-with-spring-cloud-stream-e034eee3f394

posted @ 2021-11-19 11:47 paulwong 阅读(99) | 评论 (0)编辑 收藏

2021年11月17日 #

EVEN DRIVEN - SPRING CLOUD STREAM - Error Handling

如果Function中抛出异常,系统没有配置捕获异常,则异常消息会被丢弃。通常会进行配置。

@ServiceActivator(inputChannel = "my-destination.my-group.errors")
    public void handleError(ErrorMessage message) {
        Throwable throwable = message.getPayload();
        log.error("截获异常", throwable);

        Message<?> originalMessage = message.getOriginalMessage();
        assert originalMessage != null;

        log.info("原始消息体 = {}", new String((byte[]) originalMessage.getPayload()));
    }

详情参考:
https://www.itmuch.com/spring-cloud/spring-cloud-stream-error-handling/

posted @ 2021-11-17 10:50 paulwong 阅读(85) | 评论 (0)编辑 收藏

2021年11月15日 #

EVEN DRIVEN - SPRING CLOUD STREAM - Function Component

如果要在SPRING CLOUD STREAM中和其他中间件打交道,如FILE、FTP、HTTP等,则要用到SPRING CLOUD FUNCTION。

组件地址:
https://github.com/spring-cloud/stream-applications/tree/main/functions

特殊组件,将FUNCTION变成HTTP ENDPOINTS:
https://github.com/spring-cloud/spring-cloud-function/tree/main/spring-cloud-starter-function-web
https://github.com/spring-cloud/spring-cloud-function/tree/main/spring-cloud-starter-function-webflux

posted @ 2021-11-15 17:40 paulwong 阅读(85) | 评论 (0)编辑 收藏

EVEN DRIVEN - SPRING CLOUD STREAM - Routing Function

SPRING CLOUD STREAM内置了一个RoutingFunction,能将MESSAGE路由到应用的其他FUNCTION中。
对接RoutingFunction可发送消息到其外部DESTINATION中或用“|”连接符连接。

application.yaml
# This setting can increase or decrease the rate of message production (1000 = 1s)
#
 spring.cloud.stream.poller.fixed-delay=1000
#
 DefaultPollerProperties

# This setting can control which function method in our code will be triggered if there are multiple
#
 spring.cloud.function.definition=supplyLoan

# Give the autogenerated binding a friendlier name
spring:
   application:
      name: loan-check-rabbit
   banner:
      location: classpath:/banner-rabbit.txt
   cloud:
      #BindingServiceProperties
      stream:
         #StreamFunctionProperties
         function:
            definition: loadCheckerFunction;loanCheckerDecieder;loanCheckerConsumer;\
                        loanDeclinedConsumer;loanApprovedConsumer;loanCheckerProcessor|functionRouter
            routing:
               enabled: true
         #BindingProperties
         bindings:
            loanCheckerProcessor|functionRouter-in-0:
               destination: queue.pretty.log.messages
               binder: local_rabbit
               
            loanApprovedConsumer-in-0:
               destination: load.approved
               binder: local_rabbit
            loanDeclinedConsumer-in-0:
               destination: load.declined
               binder: local_rabbit
               
            loanCheckerDecieder-in-0:
               destination: queue.pretty.log.messages.222
               binder: local_rabbit
            loanCheckerDecieder-out-0:
               destination: queue.pretty.approved.messages
               binder: local_rabbit
            loanCheckerConsumer-in-0:
               destination: queue.pretty.approved.messages
               binder: local_rabbit
         #BinderProperties
         binders:
            local_rabbit:
               type: rabbit
               environment:
                  spring:
                     rabbitmq:
                        host: 10.80.27.69
                        port: 5672
                        username: guest
                        password: guest
                        virtual-host: my-virtual-host
                        
                        
logging:
   level:
      root: info
      org.springframework:
         cloud.function: debug
         #retry: debug


LoanCheckConfiguration.java
package com.paul.testspringcloudstream.loancheck.config;

import java.util.function.Consumer;
import java.util.function.Function;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.function.context.MessageRoutingCallback;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;

import com.paul.testspringcloudstream.common.model.Loan;
import com.paul.testspringcloudstream.common.model.Status;
import com.paul.testspringcloudstream.loancheck.router.LoanCheckerRouter;
import com.paul.testspringcloudstream.loancheck.service.LoanProcessor;
import com.paul.testspringcloudstream.loancheck.service.LoanService;

@Configuration
public class LoanCheckConfiguration {
    
    private static final Logger log = LoggerFactory.getLogger(LoanCheckConfiguration.class);
    private static final Long MAX_AMOUNT = 10000L;
    private static final String LOG_PATTERN = "{} - {} {} for ${} for {}";
    
    @Autowired
    public void test(Consumer<Loan> loanCheckerConsumer) {
        log.info("{}", loanCheckerConsumer.getClass());
    }
    
    
    @Bean
    public Consumer<Loan> loanCheckerConsumer(){
        return loan -> 
            log.info(LOG_PATTERN, "loanCheckerConsumer", loan.getStatus(), loan.getUuid(), loan.getAmount(), loan.getName());
    }
    
    @Bean
    public Consumer<Loan> loanDeclinedConsumer(){
        return loan -> 
            log.info(LOG_PATTERN, "loanDeclinedConsumer", loan.getStatus(), loan.getUuid(), loan.getAmount(), loan.getName());
    }
    
    @Bean
    public Consumer<Loan> loanApprovedConsumer(){
        return loan -> 
            log.info(LOG_PATTERN, "loanApprovedConsumer", loan.getStatus(), loan.getUuid(), loan.getAmount(), loan.getName());
    }
    
    @Bean
    public MessageRoutingCallback loanCheckerRouter() {
        return new LoanCheckerRouter();
    }
    
    @Bean
    public Function<Loan, Loan> loanCheckerProcessor(
        LoanService loanService
    ){
        return loan -> loanService.check(loan);
    }
    
    @Bean
    public Function<Loan, Message<Loan>> loanCheckerProcessorBak(
        LoanService loanService
    ){
        return loan -> {
            Loan result = loanService.check(loan);
            String sendTo = Status.DECLINED.name().equals(result.getStatus()) ? 
                        LoanProcessor.DECLINED_OUT : LoanProcessor.APPROVED_OUT;
            
            return MessageBuilder.withPayload(result)
                        .setHeader("spring.cloud.stream.sendto.destination", sendTo)
                        .build();
        };
    }
    
    @Bean
    public Consumer<Loan> loanCheckerDecieder(StreamBridge streamBridge){
        return loan -> {
            log.info(LOG_PATTERN, "loanCheckerDecieder", loan.getStatus(), loan.getUuid(), loan.getAmount(), loan.getName());

            if (loan.getAmount() > MAX_AMOUNT) {
                loan.setStatus(Status.DECLINED.name());
                streamBridge.send(LoanProcessor.DECLINED_OUT, "local_rabbit", loan);
            } else {
                loan.setStatus(Status.APPROVED.name());
                streamBridge.send(LoanProcessor.APPROVED_OUT, "local_rabbit", loan);
            }

            log.info(LOG_PATTERN, "loanCheckerDecieder", loan.getStatus(), loan.getUuid(), loan.getAmount(), loan.getName());
        };
    }

}


LoanCheckerRouter.java,将路由条件统一在此处
package com.paul.testspringcloudstream.loancheck.router;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.function.context.MessageRoutingCallback;
import org.springframework.messaging.Message;

import com.paul.testspringcloudstream.common.model.Loan;
import com.paul.testspringcloudstream.common.model.Status;

public class LoanCheckerRouter implements MessageRoutingCallback{
    
    private static final Logger log = LoggerFactory.getLogger(LoanCheckerRouter.class);

    @Override
    public String functionDefinition(Message<?> message) {
        
//        byte[] resultByte = (byte[])message.getPayload();
//        String resultString = new String(resultByte);
//        
//        return "loanDeclinedConsumer";
        
        Loan result = (Loan)message.getPayload();
        
        log.info("Loan status: {}", result.getStatus());
        
        return Status.DECLINED.name().equals(result.getStatus()) ? 
                    "loanDeclinedConsumer" : "loanApprovedConsumer";
    }

}

posted @ 2021-11-15 14:46 paulwong 阅读(91) | 评论 (0)编辑 收藏

仅列出标题  下一页