消息驱动消费者

cloud-stream-rabbitmq-consumer8802

POM

<dependencies>

<dependency>

<groupId>org.springframework.boot</groupId>

<artifactId>spring-boot-starter-web</artifactId>

</dependency>

<!--监控-->

<dependency>

<groupId>org.springframework.boot</groupId>

<artifactId>spring-boot-starter-actuator</artifactId>

</dependency>

<!--eureka client-->

<dependency>

<groupId>org.springframework.cloud</groupId>

<artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>

</dependency>

<!--stream rabbit -->

<dependency>

<groupId>org.springframework.cloud</groupId>

<artifactId>spring-cloud-starter-stream-rabbit</artifactId>

</dependency>

<!--热部署-->

<dependency>

<groupId>org.springframework.boot</groupId>

<artifactId>spring-boot-devtools</artifactId>

<scope>runtime</scope>

<optional>true</optional>

</dependency>

<dependency>

<groupId>org.projectlombok</groupId>

<artifactId>lombok</artifactId>

<optional>true</optional>

</dependency>

<dependency>

<groupId>org.springframework.boot</groupId>

<artifactId>spring-boot-starter-test</artifactId>

<scope>test</scope>

</dependency>

</dependencies>

YAML

server:

port: 8802

spring:

application:

name: cloud-stream-consumer

cloud:

stream:

binders: #在此处配置要绑定的rabbitmq的服务信息

defaultRabbit: #表示定义的名称，用于binding整合

type: rabbit #消息组件类型

environment: #设置rabbitmq的相关环境配置

spring:

rabbitmq:

host: localhost

port: 5672

username: guest

password: guest

bindings: #服务的整合处理

- input: #这个名字是一个通道的名称

destination: studyExchange #表示要使用的Exchange名称定义

content-type: application/json #设置消息类型，本次为json，本文要设置为“text/plain”

binder: defaultRabbit #设置要绑定的消息服务的具体设置

# group: bsxA #绑定分组 防止消息重复消费

eureka:

client:

service-url:

defaultZone: http://localhost:7001/eureka

instance:

lease-renewal-interval-in-seconds: 2 #设置心跳的时间间隔（默认是30S)

lease-expiration-duration-in-seconds: 5 #如果超过5S间隔就注销节点 默认是90s

instance-id: receive-8802.com #在信息列表时显示主机名称

prefer-ip-address: true #访问的路径变为IP地址

主启动

@SpringBootApplication

@EnableEurekaClient

public class StreamMQMain8802 {

public static void main(String[] args) {

SpringApplication.run(StreamMQMain8802.class, args);

}

}

业务类

消费者只有 Controller

Controller

import org.springframework.beans.factory.annotation.Value;

import org.springframework.cloud.stream.annotation.EnableBinding;

import org.springframework.cloud.stream.annotation.StreamListener;

import org.springframework.cloud.stream.messaging.Sink;

import org.springframework.messaging.Message;

import org.springframework.stereotype.Component;

@Component

@EnableBinding(Sink.class)

public class ReceiverMessageListenerController {

@Value("${server.port}")

private String serverPort;

@StreamListener(Sink.INPUT)

public void input(Message<String> message) {

System.out.println("消费者1号， -----> 接受到的消息： " + message.getPayload()

+ "\t port: " + serverPort);

}

}

测试

• 启动 7001、8801、8802
• 访问 http://localhost:8801/sendMessage
• 8802 控制台可以看到从消息中间件中读取到的数据

消费者1号， -----> 接受到的消息： 26dec863-11da-43aa-ab4a-0716fce5663f port: 8802

消费者1号， -----> 接受到的消息： 9e974b57-26bf-451a-bdfe-4401c3603a30 port: 8802

消费者1号， -----> 接受到的消息： e1ab7d59-c5f8-4122-b905-9c8d3720ede5 port: 8802