spring cloud stream_大数据系统

spring cloud stream

概念：spring cloud stream 是一个集成消息中间件的框架，用于构建与共享消息传递系统连接的高度可扩展的时间驱动微服务。支持消息发布，消息订阅，消息分区处理；解决了开发人员无感知的使用消息中间件

三个核心模块：

目标绑定器：复制提供与外部消息中间件的集成
目标绑定：外部消息中间件与应用代码之间的桥梁
消息：生产者和消费者与目标绑定器进行通讯的数据结构

已经实现的目标绑定器如下：

RabbitMQ
Apache Kafka
Kafka Streams
Amazon Kinesis
Google PubSub (partner maintained)
Solace PubSub+ (partner maintained)
Azure Event Hubs (partner maintained)
AWS SQS (partner maintained)
AWS SNS (partner maintained)
Apache RocketMQ (partner maintained)

Stream 工作原理

Source ：当需要发送消息时，我们就需要通过 Source.java，它会把我们要发送的消息进行序列化，然后将数据发送到Channel中；
Sink : 监听消息需要通过Sink.java，它负责从消息通道中获取消息，并将消息序列化成消息对象，然后交给具体的消息监听处理；
Channel ：通常我们想消息中间件发送消息活监听消息时需要指定主题(Topic) 和消息队列名称。通过 Channel 对象，实现与消息中间件的通讯
Binder ：通过不通的 binder 可以实现与不通的消息中间件整合，binder 提供统一的消息收发接口，从而使得我们可以更具实际部署的消息中间件，或者更具实际生产中部署的中间件调整我们的配置

废话不多说，上代码，demo中引入的 kafka 和 rocket，因为业务上一般只使用一种类型的mq，所有这里只设置了一个input/output，在pom文件中如果同事引入了多种mq，需要指定一个default-binder

配置文件 application.yaml

server:
  port: ${random.int[10000,19999]}

spring:
  application:
    name: stream-mq-lq
  profiles:
    active: kafka
  cloud:
    stream:
      default-binder: kafka

配置文件 application-kafka.yaml

spring:
  cloud:
    stream:
      bindings:
        biz-output:
          destination: topic-x
          content-type: application/json
        biz-input:
          destination: topic-x
          content-type: application/json
          # if use group is Clustering ,else is broadcasting
          #group: test-lq
      # kafka mq config
      kafka:
        binder:
          brokers: localhost:9092

配置文件 application-rocket.yaml

spring:
  cloud:
    stream:
      bindings:
        biz-output:
          destination: MAGIC_TOPIC
          content-type: application/json
        biz-input:
          destination: MAGIC_TOPIC
          content-type: application/json
          group: mq-log
      # rocket mq config
      rocketmq:
        binder:
          name-server: localhost:9876
        bindings:
          # producer config
          biz-output:
            # RocketMQBindingProperties
            producer:
              group: mq-log
              sync: false
              maxMessageSize: 1048576
          # consumer config
          biz-input:
            # RocketMQConsumerProperties
            consumer:
              enabled: true
              # default Clustering
              broadcasting: true

pom文件



    4.0.0
    com.lq
    spring-cloud-stream-t
    0.0.1-SNAPSHOT


    
        2.3.2.RELEASE
        Hoxton.SR8
        2.2.5.RELEASE
    

    
        
            
                org.springframework.boot
                spring-boot-starter-parent
                ${spring.boot.version}
                pom
                import
            
            
                org.springframework.cloud
                spring-cloud-dependencies
                ${spring.cloud.version}
                pom
                import
            
            
                com.alibaba.cloud
                spring-cloud-alibaba-dependencies
                ${spring.cloud.alibaba.version}
                pom
                import
            
        
    
    
        
        
            org.springframework.boot
            spring-boot-starter-web
        
        
        
        
            org.springframework.cloud
            spring-cloud-stream-binder-kafka
        
        
            org.springframework.boot
            spring-boot-starter-test
            test
        
        
            org.projectlombok
            lombok
            1.18.8
        
    

    
        
            
                org.springframework.boot
                spring-boot-maven-plugin

@EnableBinding(MySink.class)
public class MessageConsumer {

    @StreamListener(MySink.INPUT)
    public void receive(@Payload Message message){
        String msg = message.getPayload();
        MessageHeaders headers = message.getHeaders();
        System.out.println("收到消息 -> "+msg+" header-> "+headers.toString());
    }
}

@EnableBinding(MySource.class)
public class MessageProducer {
    @Resource
    private MySource source;
    @Value("${spring.cloud.stream.default-binder}")
    private String mqType;

    public void sendMsg(String msg) {
        boolean send = false;
        if ("rocket".equals(mqType)) {
            send =source.bizOutput().send(MessageBuilder.withPayload(msg).setHeader("TAGS", "tag1").build());
        } else {
            send =source.bizOutput().send(MessageBuilder.withPayload(msg).build());
        }
        System.out.println("send result=" + send);
    }
}

public interface MySink {

    String INPUT = "biz-input";

    @Input("biz-input")
    SubscribableChannel bizInput();
}

public interface MySource {



    @Output("biz-output")
    MessageChannel bizOutput();
}

@Component
public class MqUtil {

    private static MessageProducer mp;
    @Resource
    private MessageProducer imp;

    @PostConstruct
    public void init(){
        MqUtil.mp=this.imp;
    }

    public static void sendMsg(String msg) {
        if (null == mp) {
            System.out.println("mq init ...");
        }
        mp.sendMsg(msg);
    }
}

@RestController
@RequestMapping("msg")
public class MsgController {

    @Resource
    private MessageProducer messageProducer;


    @GetMapping("bizMq")
    public void sendBizMq(String msg) {
        messageProducer.sendMsg(msg);
        MqUtil.sendMsg("MqUtil static send "+msg);
    }
}

spring cloud stream

大数据系统相关栏目本月热门文章