SpringBoot 集成 RabbitMQ 消息队列

SpringBoot 集成 RabbitMQ 消息队列

• 一、添加依赖
• 二、在 Application.yml 配置文件中添加配置信息
• 三、定义消息队列
• 四、配置生产者
• 五、配置消费者
• 六、编写测试类
• 七、附图

一、添加依赖

• 创建 SpringBoot 项目时，勾选 Messaging—>RabbitMQ，自动添加
• 或者在 pom.xml 文件添加依赖

<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>

二、在 Application.yml 配置文件中添加配置信息

spring:
rabbitmq:
host: ip
port: 端口
username: 用户名
password: 密码

cloud:
stream:
bindings:
input:
content-type: application/json
destination: "消息队列名"
consumer:
concurrency: 20
output:
content-type: application/json
destination: "消息队列名"

mqEnabled: true

三、定义消息队列

@Component
public interface Processor {
String INPUT  = "input";
String OUTPUT = "output";

@Input(INPUT)
SubscribableChannel input();

@Output(OUTPUT)
MessageChannel output();
}

四、配置生产者

@Service
public class Producer {

@Autowired(required = false)
private Processor Processor;

/**
* 发送至mq
* @param monitorInfo 要发送的消息
* @return
*/
public boolean sendMonitorInfo(MonitorInfo monitorInfo){
boolean sendResult = processor.output().send(new GenericMessage<>(monitorInfo));
return sendResult;
}
}

五、配置消费者

@EnableBinding(value = Processor.class)
@ConditionalOnProperty(name = "mqEnabled" ,havingValue = "true")
public class Consumer {

@Autowired
private DoSomethingService doSomethingService;

/**
* 消费
*
* @param monitorInfo
* @throws Exception
*/
@StreamListener(Processor.INPUT)
public void consume(MonitorInfo monitorInfo) {
doSomethingService.doSomething(monitorInfo);
}
}

六、编写测试类

@RunWith(SpringRunner.class)
@SpringBootTest
public class MqTest {

@Autowired
private Processor processor;

@Test
public void sendMq() throws InterruptedException {
Thread.sleep(2000);
processor.Output().send(new GenericMessage<>("111"));
Thread.sleep(2000);
}
}

七、附图

• 点赞
• 收藏
• 分享
• 文章举报