#
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
在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
如果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/
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";
}
}
SPRING CLOUD STREAM 3.x 版本时，之前的一些编程模式，如@Enablebindding，@StreamListenner等注释被废弃了，这是由于一些框架的代码必需由用户编写，如配置框架用的Input MessageChannel，Output MessageChannel，连接MessageHandler与MessageChannel等，被视为不必要的动作。为了简化用户代码，于是推出Functional Programming Model。
引入了新名词：Supplier、Function与Consumer。实际上这几个类可视为Adapter，如果之前已经有存在的Service类，且方法名为各种各样，可以重新包装成Supplier、Function与Consumer，并在固定的方法名：apply/get/accept中调用Service的方法。
Supplier
当在配置文件中注入此类型的Bean，并在spring.cloud.stream.function.definition加入此Bean的名称，SPRING CLOUD STREAM就会帮你生成一个Output MessageChannel，并连接上此Bean，后续只需要在BINDDING中加入对应的Destination Name，即可向BROKER发消息了。
Consumer
当在配置文件中注入此类型的Bean，并在spring.cloud.stream.function.definition加入此Bean的名称，SPRING CLOUD STREAM就会帮你生成一个Input MessageChannel，并连接上此Bean，后续只需要在BINDDING中加入对应的Destination Name，即可收到BROKER推送关于此Destination的消息了。
Function
当在配置文件中注入此类型的Bean，并在spring.cloud.stream.function.definition加入此Bean的名称，SPRING CLOUD STREAM就会帮你生成一个Input和Output MessageChannel，并连接上此Bean，后续只需要在BINDDING中分别对Input和Output MessageChannel加入对应的Destination Name1/Name2，即可收到BROKER推送关于此Destination的消息，也可以向BROKER发消息了。
与SPRING INTEGRATION的整合
如果要对消息进行复杂处理，如拆分消息、聚合消息、IF ELSE消息等，就要借助SPRING INTEGRATION了。
@Bean
public IntegrationFlow upperCaseFlow(LoanService loanService) {
return IntegrationFlows
//turn this IntegrationFlow as a gateway, here is a Function interface
//with loadCheckerFunction as bean name
.from(LoadCheckerFunction.class, gateway -> gateway.beanName("loadCheckerFunction"))
.handle(loanService, "check")
.logAndReply(LoggingHandler.Level.WARN);
}
public interface LoadCheckerFunction extends Function<Loan, Loan>{
}
IntegrationFlows.from(Class<?> serviceInterface)是可以将本IntegrationFlow包装成serviceInterface的实现类，如果调用此接口，最终会返回IntegrationFlow最后一个步骤的实体，如果这个serviceInterface是Function的话，刚好和SPRING CLOUD STREAM对接上。
后续在spring.cloud.stream.function.definition加入此Bean的名称loadCheckerFunction，SPRING CLOUD STREAM就会帮你生成一个Input和Output MessageChannel，并连接上此Bean，再在BINDDING中分别对Input和Output MessageChannel加入对应的Destination Name1/Name2，即可收到BROKER推送关于此Destination的消息，也可以向BROKER发消息。
application.yaml
# This setting can increase or decrease the rate of message production (1000 = 1s)
# spring.cloud.stream.poller.fixed-delay=1000
# 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:
stream:
function.definition: loadCheckerFunction
#BindingProperties
bindings:
loadCheckerFunction-in-0:
destination: queue.pretty.log.messages
binder: local_rabbit
loadCheckerFunction-out-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
Reference
https://spring.io/blog/2019/10/25/spring-cloud-stream-and-spring-integration
安装ERLANG
从这里下载0依赖的ERLANG安装包：
https://github.com/rabbitmq/erlang-rpm/releases 象这种erlang-23.3.4.8-1.el7.x86_64.rpm含el7的是CENTOS7版本，含el8的是CENTOS8版本，安装脚本
yum install -y erlang-23.3.4.8-1.el7.x86_64.rpm
安装RABBITMQ
下载地址：
https://github.com/rabbitmq/rabbitmq-server/releases安装脚本：
yum install -y erlang-23.3.4.8-1.el7.x86_64.rpm拷贝配置文件
下载配置文件样例：
https://github.com/rabbitmq/rabbitmq-server/blob/master/deps/rabbit/docs/rabbitmq.conf.example粘贴并重命名文件：
/etc/rabbitmq/rabbitmq.conf开启WEB控制台
/lib/rabbitmq/bin/rabbitmq-plugins enable rabbitmq_management
配置guest可远程访问
## Uncomment the following line if you want to allow access to the
## guest user from anywhere on the network.
loopback_users.guest = false
配置开机启动
chkconfig rabbitmq-server on
启动实例
systemctl start rabbitmq-serve
systemctl stop rabbitmq-serve
访问控制台，guest/guest
http://10.80.27.69:15672/#/
Reference
https://www.cnblogs.com/ZhuChangwu/p/14093107.htmlhttps://juejin.cn/post/6933040530519506957
通常微服务应用之间的通信是通过HTTP调用，吞吐性不建都高，高并发的场景建议使用EVENT DRIVEN的框架，即使用MESSAGE通信。
即A微服务应用将数据发送到MESSAGE BROKER中的某个DESTINATION，此DESTINATION是广播型，非点对点型。B微服务应用订阅此DESTINATION，当有新MESSAGE到达此DESTINATION时，MESSAGE BROKER会将此MESSAGE推送给B应用。所有对此MESSAGE有需要的应用均可订阅，从而收到此MESSAGE。
SPRING CLOUD 中EVENT DRIVEN的框架就是SPRING CLOUD STREAM。其底层是使用SPRING INTEGRATION实现。
SPRING CLOUD STREAM有以下新名词：
是对MESSAGE BROKER操作方法的抽象，即应用通过此BINDER操作MESSAGE BROKER。目前只实现了RABITMQ和KAFKA。
MESSAGE从SPRING CLOUD STREAM传给应用或相反是通过CHANNEL传递的，这点和SPRING INTEGRATION是一样的。
MESSAGE从应用传给SPRING CLOUD STREAM的CHANNEL，叫@INPUT，包含这种CHANNEL的接口叫SOURCE。
MESSAGE从SPRING CLOUD STREAM传给应用的CHANNEL，叫@OUPUT，包含这种CHANNEL的接口叫SINK。
绑定哪个@INPUT或哪个@OUPUT与哪个DESTINATION发送或接收关系的MAPPING。
应用启动时就会建立EnableBinding指定的接口中的CHANNEL
默认下如果同一个应用部署了多个实例，则每个实例都会收到MESSAGE，这时如果设置了消费者群组名称，则同一个名称下的多个实例，只有一个能收到MESSAGE。
如果为MESSAGE指定规则，如MESSAGE某个字段值以A开头为一个规则，以B开头为一个规则，那么以A开头的MESSAGE会放到同一个分区中。
这样使用就很简单了，只要取得OUTPUT CHANNEL，就可以发送MESSAGE，将代码关联到INPUT CHANNEL，就能在收到MESSAGE时，相关代码就会被执行。