Spring Cloud Stream与Kafka（二）

Spring Cloud Stream与Kafka（二）

Spring Cloud Stream提供的信道

1. Source接口

package org.springframework.cloud.stream.messaging;import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;//具有一个输出通道的可绑定接口
public interface Source {//输出通道的名称String OUTPUT = "output";//输出通道@Output(Source.OUTPUT)MessageChannel output();
}

2. Sink接口

package org.springframework.cloud.stream.messaging;import org.springframework.cloud.stream.annotation.Input;
import org.springframework.messaging.SubscribableChannel;//具有一个输入通道的可绑定接口
public interface Sink {//输入通道的名称String INPUT = "input";//输入通道@Input(Sink.INPUT)SubscribableChannel input();
}

3. Processor接口

package org.springframework.cloud.stream.messaging;//具有一个输入和一个输出通道的可绑定接口
public interface Processor extends Source, Sink {}

自定义Binding声明接口

1. 创建自定义绑定接口，定义@Input和@Output时如果没有名称，默认获取当前方法的名称作为绑定名称。应用的时候和其他的绑定接口一样通过@EnableBinding进行声明。

public interface CustomBinding {String INPUT1 = "input1";String OUTPUT1 = "output1";@InputSubscribableChannel input1();@OutputMessageChannel output1();
}

2. 应用自定义接口

@SpringBootApplication
@EnableBinding({CustomBinding.class, Source.class})
public class Application implements CommandLineRunner{}

Spring Cloud Stream注解

1. @Output注解指示框架将会创建一个输出绑定目标。

public @interface Output {//指定绑定目标名称及绑定目标的Bean名称，以及作为默认的destination名称String value() default "";
}

2. @Input注解指示框架将会创建一个输入绑定目标。

public @interface Input {//指定绑定目标名称及绑定目标的Bean名称，以及作为默认的destination名称   String value() default "";
}

3. @StreamListener注解可以把方法标记为通过@EnableBinding注解声明的输入的监听器。

@Target({ ElementType.METHOD, ElementType.ANNOTATION_TYPE })
@Retention(RetentionPolicy.RUNTIME)
@MessageMapping
@Documented
public @interface StreamListener {//方法订阅的绑定目标的名称比如通道@AliasFor("target")String value() default "";@AliasFor("value")String target() default "";//分派给此方法的所有项都必须要满足的条件String condition() default "";//默认是true，当有一个@SendTo注解时，把流入的头信息复制到流出的消息中String copyHeaders() default "true";
}

4. @SendTo注解指示方法的返回值被转换为消息发送到指定的目的地。

public @interface SendTo {//从方法的返回值创建的消息的目的地String[] value() default {};
}

5. @EnableBinding根据作为值传递给注释的接口列表，启用带有@Input和@Output注释的目标绑定到代理。

@Target({ ElementType.TYPE, ElementType.ANNOTATION_TYPE })
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Inherited
@Configuration
@Import({ BindingBeansRegistrar.class, BinderFactoryConfiguration.class })
@EnableIntegration
public @interface EnableBinding {//带有@Input或@Output注解的方法的接口列表表示绑定目标Class<?>[] value() default {};
}

package org.springframework.cloud.stream.config;import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.binding.BindingBeanDefinitionRegistryUtils;
import org.springframework.context.annotation.ImportBeanDefinitionRegistrar;
import org.springframework.core.annotation.AnnotatedElementUtils;
import org.springframework.core.annotation.AnnotationAttributes;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.core.type.AnnotationMetadata;
import org.springframework.util.ClassUtils;//绑定Bean注册器
public class BindingBeansRegistrar implements ImportBeanDefinitionRegistrar {@Overridepublic void registerBeanDefinitions(AnnotationMetadata metadata,BeanDefinitionRegistry registry) {AnnotationAttributes attrs = AnnotatedElementUtils.getMergedAnnotationAttributes(ClassUtils.resolveClassName(metadata.getClassName(), null),EnableBinding.class);for (Class<?> type : collectClasses(attrs, metadata.getClassName())) {if (!registry.containsBeanDefinition(type.getName())) {BindingBeanDefinitionRegistryUtils.registerBindingTargetBeanDefinitions(type, type.getName(), registry);BindingBeanDefinitionRegistryUtils.registerBindingTargetsQualifiedBeanDefinitions(ClassUtils.resolveClassName(metadata.getClassName(), null), type,	registry);				}}}private Class<?>[] collectClasses(AnnotationAttributes attrs, String className) {EnableBinding enableBinding = AnnotationUtils.synthesizeAnnotation(attrs,EnableBinding.class, ClassUtils.resolveClassName(className, null));return enableBinding.value();}}

发布与订阅

1. Spring Cloud Stream默认在接收和发送消息时对应的消息格式类型都是JSON，我们可以通过绑定的contentType属性进行指定。当发送和接收消息时都会被MessageConverter消息转换器进行转换。

spring.cloud.stream.bindings.output.content-type=application/json
# 发布者配置
spring.cloud.stream.bindings.output1.destination=test
# 消费者配置
spring.cloud.stream.bindings.input1.destination=test
spring.cloud.stream.bindings.input1.group=test_group