cloud stream 官方文档阅读笔记4

矫情吗；* 2022-04-01 08:39 287阅读 0赞

5、编程模型  
要了解编程模型，您应该熟悉以下核心概念：  
Destination Binders 负责提供与外部消息系统集成的组件  
Destination Bindings 外部消息传递系统和应用程序之间的桥梁，提供了消息的生产者和消费者  
Message：生产者和消费者用于与目标绑定器（以及通过外部消息传递系统的其他应用程序）通信的规范数据结构。

5.1、Destination Binders  
目标绑定器是Spring Cloud Stream的扩展组件，负责提供必要的配置和实现，以促进与外部邮件系统的集成，  
此集成负责与生产者和使用者之间的消息的连接，委派和路由，数据类型转换，用户代码调用等  
它处理许多锅炉板的责任，否则它们会落在你的肩膀上，然而，为了实现这一点，绑定器仍然需  
要来自用户的简约但需要的指令集的形式的一些帮助，其通常以某种类型的配置的形式出现。

5.2、Destination Bindings  
如前所述，Destination Bindings在外部消息传递系统和应用程序提供的生产者和消费者之间提供了一个桥梁。  
将@EnableBinding 注解应用于其中一个应用程序的配置类可定义目标绑定。  
@EnableBinding注释本身使用@Configuration进行元注释，并触发Spring Cloud Stream基础结构的配置。  
以下示例显示了一个完全配置且正常运行的Spring Cloud Stream应用程序，该应用程序将作为String类型  
从INPUT目标接收消息的有效负载（请参阅内容类型协商部分），将其记录到控制台并将其转换为大写后将其  
发送到OUTPUT目标。

@SpringBootApplication
    @EnableBinding(Processor.class)
    public class MyApplication{ 
      public static void main(String[] args){ 
        SpringApplication.run(MyApplication.class, args);
      }
      @StreamListener(Processor.INPUT)
      @SendTo(Processor.OUTPUT)
      public String handle(String value){ 
        System.out.println("Received: " + value);
        return value.toUpperCase();
      }
    }

如您所见，@ EnableBinding批注可以将一个或多个接口类作为参数.这些参数称为绑定，它们包含表示可绑定组件的方法.  
这些组件通常是基于通道的绑定器（例如Rabbit，Kafka等）的消息通道（请参阅Spring Messaging）。  
然而，其他类型的绑定可以为相应技术的本机特征提供支持。例如，Kafka Streams binder（以前称为KStream）允许直  
接绑定到Kafka Streams的本机绑定（有关详细信息，请参阅Kafka Streams）。  
Spring Cloud Stream已经为典型的消息交换合同提供了绑定接口，其中包括:  
Sink: 通过提供消息使用的目标来标识消息使用者的接口  
Source: 通过提供生成的消息发送到的目标来标识消息生产者的接口  
Processor: 通过暴露允许消费和生成消息的两个目标来封装接收器和源合同  
上面三个接口的实现代码：

public interface Sink{ 
    
      String INPUT = "input";
    
      @Input(Sink.INPUT)
      SubscribableChannel input();
    }
    
    public interface Source{ 
    
      String OUTPUT = "output";
    
      @Output(Source.OUTPUT)
      MessageChannel output();
    }
    
    public interface Processor extends Source, Sink{ }

虽然前面的示例满足大多数情况，但您也可以通过定义自己的绑定接口来定义自己的合同，  
并使用@Input和@Output注释来标识实际的可绑定组件。  
例如：

public interface Barista { 
    
      @Input
      SubscribableChannel orders();
    
      @Output
      MessageChannel hotDrinks();
    
      @Output
      MessageChannel coldDrinks();
    }

使用前面示例中显示的接口作为@EnableBinding的参数，分别触发创建名为orders，hotDrinks和  
coldDrinks的三个绑定通道。  
您可以根据需要提供尽可能多的绑定接口，作为@EnableBinding批注的参数，如以下示例所示：

@EnableBinding(value = { Order.class, Paymet.class})

在Spring Cloud Stream中，可绑定的MessageChannel组件是Spring Messaging MessageChannel（用于出站）  
及其扩展，SubscribableChannel（用于入站）

#### 可轮询目的地绑定 ####

虽然之前描述的绑定支持基于事件的消息消耗，但有时您需要更多控制，例如消耗率,从2.0版开始，  
您现在可以绑定可轮询的使用者,以下示例显示如何绑定可轮询使用者:

public interface PolledBar{ 
    
      @Input
      PollableMessageSource orders();
    
      . . .
    }

在这种情况下，PollableMessageSource的实现绑定到订单“channel”。

#### 自定义频道名称 ####

通过使用@Input和@Output注释，您可以为通道指定自定义通道名称，如以下示例所示

public interface Barista{ 
      @Input("inboundOrders")
      SubscribableChannel orders();
    }

在前面的示例中，创建的绑定通道名为inboundOrders  
通常，您不需要直接访问单个通道或绑定（除了通过@EnableBinding注释配置它们）  
但是，有时可能会出现测试或其他特殊情况

除了为每个绑定生成通道并将它们注册为Spring bean之外，对于每个绑定的接口，Spring Cloud  
Stream生成实现接口的bean，这意味着您可以通过在应用程序中自动连接来访问表示绑定或单个通  
道的接口，如以下两个示例所示：

A u t o w i r e B i n d i n g i n t e r f a c e Autowire Binding interface AutowireBindinginterface

@Autowire
    private Source Source;
    
    public void sayHello(String name){ 
      Source.output().send(MessageBuilder.withPayload(name).build());
    }

A u t o w i r e i n d i v i d u a l c h a n n e l Autowire individual channel Autowireindividualchannel

@Autowire
    private MessageChannel output;
    
    public void sayHello(String name){ 
      output.send(MessageBuilder.withPayload(name).build());
    }

对于自定义通道名称的情况或需要特定命名通道的多通道方案，您还可以使用标准Spring的@Qualifier注释。  
以下示例显示如何以这种方式使用@Qualifier注释：

@Autowire
    @Qualifier("myChannel")
    private MessageChannel output;

5.3、生产和消费消息  
您可以使用Spring Integration注释或Spring Cloud Stream本机注释编写Spring Cloud Stream应用程序。

5.3.1、Spring集成支持  
Spring Cloud Stream建立在Enterprise Integration Patterns定义的概念和模式之上，并依赖于已经建立的内部实现  
和Spring项目组合中的企业集成模式的流行实现：Spring Integration框架。  
所以它唯一的理由就是支持Spring Integration已经建立的基础，语义和配置选项

例如，您可以将Source的输出通道附加到MessageSource并使用熟悉的@InboundChannelAdapter注释，如下所示：

@EnableBinding(Source.class)
    public class TimeSource { 
      @Bean 
      @InboundChannelAdapter(value = Source.OUTPUT, poller = @Poller(fixedDelay = "10", 
      maxMessagesPerPoll = "1"))
      public MessageSource<String> timerMessageSource(){ 
        return () -> new GenericMessage<>("Hello Spring Cloud Stream");
      }
    }

同样，您可以使用@Transformer或@ServiceActivator，同时为处理器绑定契约提供消息处理程序方法的实现，  
如以下示例所示：

@EnableBinding(Processor.class)
    public class TransformerProcessor{ 
      @Transformer(inputChannel = Processor.INPUT, outputchannel = Processor.OUTPUT)
      public Object Transform(String message){ 
        return message.toUpperCase();
      }
    }

5.3.2、使用 @StreamListener注解

作为Spring Integration支持的补充，Spring Cloud Stream提供了自己的@StreamListener注解，  
以其他Spring Messaging注释（@ MessessMapping，@ JamsListener，@ RackListener等）为模型，  
并提供诸如基于内容的路由等方便的便利。

@EnableBinding(Sink.class)
    public class VoteHandler{ 
      @Autowired
      VotingService votingservice;
    
      @StreamListener(Sink.INPUT)
      public void handle(Vote vote){ 
        VotingService.record(vote);
      }
    }

与其他Spring Messaging方法一样，方法参数可以使用@Payload，@ Headers和@Header进行注释。  
对于返回数据的方法，必须使用@SendTo批注指定方法返回的数据的输出绑定目标，如以下示例所示:

@EnableBinding(Processor.class)
    public class TransformProcessor{ 
      @Autowired
      VotingService votingservice;
    
      @StreamListener(Processor.INPUT)
      @SendTo(Processor.OUTPUT)
      public VoteResult handle(Vote vote){ 
        return votingservice.record(vote);
      }
    }

5.3.3、使用@StreamListener进行基于内容的路由  
Spring Cloud Stream支持根据条件将消息分派给使用@StreamListener注释的多个处理程序方法。  
为了有资格支持条件分派，方法必须满足以下条件：  
它不能有返回值。  
它必须是单独的消息处理方法（不支持反应API方法）。

条件由注释的condition参数中的SpEL表达式指定，并针对每条消息进行评估。条件匹配的所有处理  
程序都在同一个线程中调用，并且不必假设调用发生的顺序。  
在以下带有调度条件的@StreamListener示例中，带有值为bogey的头类型的所有消息都将被调度到receiveBogey方法  
带有值为bacall的头类型的所有消息都将被分派到receiveBacall方法。

@EnableBinding(Sink.class)
    @EnableAutoConfiguration
    public static class TestPojoWithAnnotatedArguments { 
    
        @StreamListener(target = Sink.INPUT, condition = "headers['type']=='bogey'")
        public void receiveBogey(@Payload BogeyPojo bogeyPojo) { 
           // handle the message
        }
    
        @StreamListener(target = Sink.INPUT, condition = "headers['type']=='bacall'")
        public void receiveBacall(@Payload BacallPojo bacallPojo) { 
           // handle the message
        }
    }

##### 条件语境中的内容类型谈判 #####

使用@StreamListener的condition参数了解基于内容的路由背后的一些机制非常重要,特别是在整个消息类型的背景下。  
如果您在继续操作之前熟悉内容类型协商，这也可能有所帮助。  
请考虑以下情形：

@EnableBinding(Sink.class)
    @EnableAutoConfiguration
    public static class CatsAndDogs { 
    
        @StreamListener(target = Sink.INPUT, condition = "payload.class.simpleName=='Dog'")
        public void bark(Dog dog) { 
           // handle the message
        }
    
        @StreamListener(target = Sink.INPUT, condition = "payload.class.simpleName=='Cat'")
        public void purr(Cat cat) { 
           // handle the message
        }
    }

上述代码完全有效。它编译和部署没有任何问题，但它永远不会产生您期望的结果。

那是因为你正在测试你期望的状态中尚不存在的东西。这是因为消息的有效载荷尚未从有线格式（byte \[\]）  
转换为所需类型。换一种说法,它还没有经过内容类型协商中描述的类型转换过程。  
因此，除非使用评估原始数据的SPeL表达式（例如，字节数组中第一个字节的值）,使用基于消息头的表达式  
（例如condition =“headers \[‘type’\] ==‘dog’”）

5.3.4、使用轮询的消费者  
使用轮询的使用者时，您可以根据需要轮询PollableMessageSource。考虑以下轮询消费者的示例：

public interface PolledConsumer { 
    
        @Input
        PollableMessageSource destIn();
    
        @Output
        MessageChannel destOut();
    
    }

鉴于前面示例中的轮询消费者，您可以按如下方式使用它

@Bean
    public ApplicationRunner poller(PollableMessageSource destIn, MessageChannel destOut) { 
        return args -> { 
            while (someCondition()) { 
                try { 
                    if (!destIn.poll(m -> { 
                        String newPayload = ((String) m.getPayload()).toUpperCase();
                        destOut.send(new GenericMessage<>(newPayload));
                    })) { 
                        Thread.sleep(1000);
                    }
                }
                catch (Exception e) { 
                    // handle failure (throw an exception to reject the message);
                }
            }
        };
    }

PollableMessageSource.poll（）方法接受MessageHandler参数（通常是lambda表达式，如此处所示）。  
如果收到并成功处理了消息，则返回true。

与消息驱动的使用者一样，如果MessageHandler抛出异常，则会将消息发布到错误通道，  
如“\[binder-error-channels\]”中所述。  
通常，poll（）方法在MessageHandler退出时确认消息。如果方法异常退出，则拒绝该消息（不重新排队）  
您可以通过重写该方法覆盖该行为，如以下示例所示

@Bean
    public ApplicationRunner poller(PollableMessageSource dest1In, MessageChannel dest2Out) { 
        return args -> { 
            while (someCondition()) { 
                if (!dest1In.poll(m -> { 
                    StaticMessageHeaderAccessor.getAcknowledgmentCallback(m).noAutoAck();
                    // e.g. hand off to another thread which can perform the ack
                    // or acknowledge(Status.REQUEUE)
    
                })) { 
                    Thread.sleep(1000);
                }
            }
        };
    }

还有一个重载的poll方法，其定义如下：

poll(MessageHandler handler, ParameterizedTypeReference<?> type)

该类型是转换提示，允许转换传入的消息有效内容，如以下示例所示：

boolean result = pollableSource.poll(received -> { 
    			Map<String, Foo> payload = (Map<String, Foo>) received.getPayload();
                ...
    
    		}, new ParameterizedTypeReference<Map<String, Foo>>() { });