springboot集成RabbitMQ非常简单,如果只是简单的使用配置非常少,springboot提供了spring-boot-starter-amqp项目对消息各种支持。
资源代码:练习用的代码。
https://github.com/xiaozhuanfeng?tab=repositories
简单使用
1、配置pom包,主要是添加spring-boot-starter-amqp的支持
<!-- RabbitMq -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
2、配置文件(生产者,消费者都配置)
配置rabbitmq的安装地址、端口以及账户信息
#RabbitMq
spring.application.name=Spring-boot-rabbitmq
spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=eric
spring.rabbitmq.password=eric
3、对列配置(生产者配置)
package com.example.demo.rabbitMq.demo;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitConfig {
@Bean
public Queue Queue() {
return new Queue("mesgQue");
}
}
注意:new Queue("String"),源码:
durable 持久化
exclusive 独占,只能被一个Consumer的conn使用
auto_delete 自动删除,在最后一个consumer断开连接后自动删除
- durable属性和auto-delete****属性可以同时生效;
- durable属性和exclusive属性会有性质上的冲突,两者同时设置时,仅exclusive****属性生效;
- auto_delete属性和exclusive****属性可以同时生效;
4、生产者:在工程rabbitMqProj1编写:
package com.example.demo.rabbitMq.demo;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@Component
public class HelloSender2 {
@Autowired
private AmqpTemplate rabbitTemplate;
public void send(int i) {
String context = "hello world....." + i;
System.out.println("Sender : " + context);
this.rabbitTemplate.convertAndSend("mesgQue", context);
}
}
5、消费者:在工程rabbitMqProj2编写:
package com.example.demo.rabbitMq.demo;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
@RabbitListener(queues = "mesgQue")
public class HelloReceiver {
@RabbitHandler
public void process(String hello) {
System.out.println("Receiver1 : " + hello);
}
}
package com.example.demo.rabbitMq.demo;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
@RabbitListener(queues = "mesgQue")
public class HelloReceiver2 {
@RabbitHandler
public void process(String hello) {
System.out.println("Receiver2 : " + hello);
}
}
注意:参数类型一定要与生产者发送的类型一致,否则当生产者消息放入队列时,消费者工程将报异常,No method found for class com.example.demo.dto.User
6、测试:
6.1、生产者
@Test
public void oneSenderToManyRec() throws Exception {
for (int i=0;i<100;i++){
helloSender2.send(i);
}
}
执行后,在RabbitMQ控制台查看,如图:
此时或许会报错:Socket Closed,需要设置virtual hosts,因为eric现在有登录后台的权限,但是没有管理队列的权限。
解决:
具体参照:https://blog.csdn.net/wabiaozia/article/details/53791366
Virtual Hosts管理:
RabbitMQ中可以虚拟消息服务器VirtualHost,每个VirtualHost相当月一个相对独立的RabbitMQ服务器,每个VirtualHost之间是相互隔离的。exchange、queue、message不能互通。 相当于mysql的db。
所以可以在控制台新增一个virtual host :eric,一个队列,注意用户时eric,注意这里需要重新指定虚拟主机:
spring.rabbitmq.virtualHost=eric
然后调整下代码中的队列名,试下,是可以正常执行的。。。
6.2、启动消费端,发现可以接收到数据,再看控制台,此时队列数据已经消费了。