springcloud-stream之消费者模块

　　现在来构建消费者模块，发送消息到 rabbitmq

　　1. 创建模块

　　2. 添加依赖

<dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
        </dependency>
        <!--   主要加入这个依赖     -->
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-devtools</artifactId>
            <scope>runtime</scope>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>cn.aib.springcloud</groupId>
            <artifactId>springclud-api-common</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

　　3. 改配置

server:
  port: 8802
spring:
  application:
    name: cloud-stream-consumer
  cloud:
    stream:
      binders:  
        defaultRabbit: 
          type: rabbit #消息组件类型
          environment: #环境配置
            spring:
              rabbitmq:
                host: localhost
                port: 5672
                username: guest
                password: guest
      bindings: # 服务的整合处理
        input: # 这次是input通道，也表示本模块是消费者模块
          destination: studyExchang # 表示要使用的Exchang名称定义
          content-type: application/json # 设置消息类型，本次为json
          binder: defaultRabbit #设置要绑定的消息服务的具体设置
          #group: atguiguA
eureka:
  client:
    service-url:
      defaultZone: http://localhost:7001/eureka
  instance:
    lease-renewal-interval-in-seconds: 2
    lease-expiration-duration-in-seconds: 5
    instance-id: receive-8802.com
    prefer-ip-address: true

　　4. 主启动

@SpringBootApplication
public class StreamConsumerApplication {
    public static void main(String[] args) {
        SpringApplication.run(StreamConsumerApplication.class, args);
    }
}

　　5. 写业务

　　　　controller：

@Component
@EnableBinding(Sink.class) //输出通道就指定为source，输入通道就指定sink
public class ReceiveMessageListenerController {

    @Value("${server.port}")
    private String serverPort;

    @StreamListener(Sink.INPUT) //监听sink通道，当通道上有消息时，就会调用下面的方法
    public void input(Message<String> message) {
        System.out.println("消费者1号，------>接收到的消息："+message.getPayload()+"	 port" +serverPort); //getPayLoad（）表示拿到消息
    }

}

　　6. 测试；只要启动即可，就会一直监听通道。当然了，稍微底层一点还没了解过，不过看架构图就大概知道流程是怎样的了