一、依赖关系
创建的 SpringBoot 模块中，引入了 amqp 的启动器
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
amqp中包括了下面两个依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-messaging</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit</artifactId>
</dependency>
二、自动配置
1、自动配置类：RabbitAutoConfiguration
2、自动配置了连接工厂 CachingConnectionFactory（通过RabbitProperties创建连接工厂）
3、RabbitProperties 封装了 Rabbit 的配置
4、RabbitTemplate 给 RabbitMQ 发送和接受消息
5、AmqpAdmin ：RabbitMQ 系统管理功能组件
三、测试
1、配置 RabbitMQ 的信息
spring.rabbitmq.host=/192.168.1.6
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
# 默认就是5672
spring.rabbitmq.port=5672
#spring.rabbitmq.virtual-host
2、测试单播
（1）发送数据
/**
* 1. 单播（点对点）
*/
@Test
public void testDirect() {
//message 需要自己构造一个，定义消息体内容和消息头
//rabbitTemplate.send(exchage, routeKey, message);
//Object 只需要传入要发送的对象，自动序列化发送给 rabbitmq
//rabbitTemplate.convertAndSend(exchage, routeKey, Object);
Map<String, Object> map = new HashMap<>();
map.put("msg", "这是第一个消息");
map.put("data", Arrays.asList("helloworld", 123, true));
//对象被默认序列化以后发送出去
rabbitTemplate.convertAndSend("exchange.direct", "njf.news", map);
}
（2）接收数据
//接收数据
@Test
public void receiveMsg() {
Object o = rabbitTemplate.receiveAndConvert("njf.news");
System.out.println(o.getClass());
System.out.println("o = " + o);
}
发送数据，将数据自动转化为 JSON 发送出去，可以自定义序列化器
3、自定义序列化器
在 RabbitTemplate 类中有一个 MessageConverter 消息转换器：
如果没有配置消息转换器就使用默认的转换器。
在SimpleMessageConverter类中的 fromMessage 方法中：
这是默认的消息转换器。
RabbitMQ 中的 MessageConverter 是消息转换接口，提供了多个实现类：
只需要在容器中配置想使用的转换器即可。
@Configuration
public class MyAMQPConfig {
@Bean
public MessageConverter messageConverter() {
return new Jackson2JsonMessageConverter();
}
}
再次进行测试，发现就可以把数据转成 JSON格式的数据了。
4、测试广播
/**
* 2、广播
*/
@Test
public void fanout() {
rabbitTemplate.convertAndSend("exchange.fanout", "", new Book("三国演义", "罗贯中"));
}
从管理平台也可以看到消息。
四、监听消息
1、@EnableRabbit 开启基于注解的 RabbitMQ 模式
@EnableRabbit //开启基于注解的 RabbitMQ 模式
@SpringBootApplication
public class SpringBoot10AmqpApplication {
public static void main(String[] args) {
SpringApplication.run(SpringBoot10AmqpApplication.class, args);
}
}
2、@RabbitListener 监听消息队列的内容
@Service
public class BookService {
@RabbitListener(queues = {"njf.news"})
public void receive(Book book){
System.out.println("book = " + book);
System.out.println("收到消息 = " + book);
}
@RabbitListener(queues = {"njf"})
public void receiveMsg(Message message){
System.out.println("收到消息 = " + message);
System.out.println(message.getBody());
System.out.println(message.getMessageProperties());
}
}
当监听到指定队列里面有内容时，就会来执行方法。
五、AmqpAdmin
AmqpAdmin ：RabbitMQ 系统管理功能组件。
AmqpAdmin：创建和删除 Queue，Exchange，Binding规则等。
示例：
@Autowired
AmqpAdmin amqpAdmin;
@Test
public void createExchange() {
//declare 开头的用于创建组件
//创建交换器
Exchange exchange = new DirectExchange("amqpAdmin.exchange");
amqpAdmin.declareExchange(exchange);
System.out.println(exchange + "创建完成");
//创建队列
amqpAdmin.declareQueue(new Queue("adqpAdmin.queue", true));
//创建绑定规则
amqpAdmin.declareBinding(new Binding("adqpAdmin.queue", Binding.DestinationType.QUEUE,
"amqpAdmin.exchange", "amqp.hhh", null));
}
效果：