Spring Cloud Stream + Kafka

以你之姓@ 2022-06-12 08:29 146阅读 0赞

> SpringCloudStream是一个构建消息驱动的微服务框架。SpringCloudStream构建在SpringBoot之上用以创建工业级的应用程序，并且Spring Integration提供了和消息代理的连接。SpringCloudStream提供几个厂商消息中间件个性化配置，引入发布订阅、消费组和分区的语义概念。添加@EnableBinding注解在你的程序中，被@StreamListener修饰的方法可以立即连接到消息代理接收流处理事件。

### **原生Spring Integration支持** ###

#### 生产者： ####

package original;
    
    import java.util.UUID;
    
    import org.springframework.cloud.stream.annotation.EnableBinding;
    import org.springframework.cloud.stream.messaging.Source;
    import org.springframework.context.annotation.Bean;
    import org.springframework.integration.annotation.InboundChannelAdapter;
    import org.springframework.integration.annotation.Poller;
    import org.springframework.integration.core.MessageSource;
    import org.springframework.messaging.support.MessageBuilder;
    
    import bean.TimeInfo;
    
    @EnableBinding(Source.class)
    public class DataProducer {
    
    
        @Bean
        @InboundChannelAdapter(value = Source.OUTPUT, poller = @Poller(fixedDelay = "10000", maxMessagesPerPoll = "1"))
        public MessageSource<TimeInfo> timerMessageSource() {
            return () -> MessageBuilder.withPayload(
                    TimeInfo.builder().time(Long.toString(System.currentTimeMillis())).label(UUID.randomUUID().toString()).build())
                    .build();
        }
    
    
    }

#### 消费者 ####

package original;
    
    import org.springframework.cloud.stream.annotation.EnableBinding;
    import org.springframework.cloud.stream.annotation.StreamListener;
    import org.springframework.cloud.stream.messaging.Sink;
    
    import bean.SinkTimeInfo;
    
    @EnableBinding(Sink.class)
    public class DataConsumer {
    
        @StreamListener(Sink.INPUT)
        public void loggerSink(SinkTimeInfo sinkTimeInfo) {
            System.out.println("Received: " + sinkTimeInfo.toString());
        }  
    
    }

### **自定义input和output通道** ###

#### output通道 ####

package custom;
    
    import org.springframework.cloud.stream.annotation.Output;
    import org.springframework.messaging.MessageChannel;
    
    public interface ProducerSource {
    
        String OUTPUT = "output";
    
        @Output(ProducerSource.OUTPUT)
        MessageChannel output();
    
    }

#### 生产者 ####

package custom;
    
    import java.util.Random;
    import java.util.UUID;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.cloud.stream.annotation.EnableBinding;
    import org.springframework.messaging.support.MessageBuilder;
    import org.springframework.scheduling.annotation.Scheduled;
    import org.springframework.stereotype.Component;
    
    import bean.Order;
    
    @EnableBinding(ProducerSource.class)
    @Component
    public class Producer {
    
        @Autowired
        private ProducerSource source;
    
        @Scheduled(fixedRate = 3000)
        public void produceHotDrinks() {
            try {
                source.output().send(MessageBuilder
                        .withPayload(Order.builder().flag("Hot").num(new Random().nextInt(100)).build()).build());
            } catch (Exception e) {
                System.out.println(e);
            }
        }
    
        @Scheduled(fixedRate = 3000)
        public void produceColdDrinks() {
            source.output().send(MessageBuilder
                    .withPayload(Order.builder().flag("Cold").num(new Random().nextInt(100)).build()).build());
        }

#### input通道 ####

package custom;
    
    import org.springframework.cloud.stream.annotation.Input;
    import org.springframework.messaging.SubscribableChannel;
    
    public interface ConsumerSink {
    
        String INPUT = "input";
        @Input(ConsumerSink.INPUT)
        SubscribableChannel input();
    }

#### 消费者 ####

package custom;
    
    import org.springframework.cloud.stream.annotation.EnableBinding;
    import org.springframework.cloud.stream.annotation.StreamListener;
    
    import bean.Order;
    
    @EnableBinding(ConsumerSink.class)
    public class Consumer {
    
        @StreamListener(ConsumerSink.INPUT)
        public void listen(Order order) {
            System.out.println("Order Received: " + order);
    
        }

### **kafka配置** ###

--- spring:
     cloud:
     stream:
     bindings:
     output:
     destination: wd
     content-type: application/json
     input:
     destination: wd
     content-type: application/json
    
     kafka:
     binder:
     zkNodes: 192.168.164.129
     brokers: 192.168.164.129
     input:
     consumer:
     resetOffsets: true ---
    spring:
     cloud:
     stream:
     bindings:
     input:
     destination: wd
     content-type: application/json
    
     kafka:
     binder:
     zkNodes: 192.168.164.129
     brokers: 192.168.164.129
     input:
     consumer:
     resetOffsets: true

[https://blog.codecentric.de/en/2016/04/event-driven-microservices-spring-cloud-stream/][https_blog.codecentric.de_en_2016_04_event-driven-microservices-spring-cloud-stream]

[http://spring-cloud.io/reference/stream/\#][http_spring-cloud.io_reference_stream]

[https_blog.codecentric.de_en_2016_04_event-driven-microservices-spring-cloud-stream]: https://blog.codecentric.de/en/2016/04/event-driven-microservices-spring-cloud-stream/
[http_spring-cloud.io_reference_stream]: http://spring-cloud.io/reference/stream/#