Spring的RabbitAdmin使用
RabbitAdmin类,可以让我们能很方便的操作队列,交换机。如:声明交换机,声明队列,绑定,清除消息,发送消息。而且可以注入到Spring的容器中,我们使用的时候直接@Autowired就可以使用。使用方式:配置文件和依赖tips:高版本spring-boot-starter-amqp我在测试时候报错,所以用了1.5版本springboot<?xml version="1.0" enco
RabbitAdmin类,可以让我们能很方便的操作队列,交换机。如:声明交换机,声明队列,绑定,清除消息,发送消息。而且可以注入到Spring的容器中,我们使用的时候直接@Autowired
就可以使用。
使用方式:
- 配置文件和依赖
tips:高版本spring-boot-starter-amqp我在测试时候报错,所以用了1.5版本springboot
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>1.5.3.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.example</groupId>
<artifactId>demo1</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>demo1</name>
<description>Demo project for Spring Boot</description>
<properties>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>3.6.5</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
#自己mq地址记得改
mq.address: 192.168.3.8:5672
mq.username: guest
mq.password: guest
mq.virtualhost: /
- 初始化RabbitAdmin的Bean
package com.example.demo1.config;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* @author:luzaichun
* @Date:2020/12/30
* @Time:21:23
**/
@Configuration
public class RabbitConfig {
@Value("${mq.address}")
private String address;
@Value("${mq.username}")
private String username;
@Value("${mq.password}")
private String password;
@Value("${mq.virtualhost}")
private String virtualhost;
@Bean
public ConnectionFactory connectionFactory(){
CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
connectionFactory.setAddresses(address);
connectionFactory.setUsername(username);
connectionFactory.setPassword(password);
connectionFactory.setVirtualHost(virtualhost);
return connectionFactory;
}
@Bean
public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory){
RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
rabbitAdmin.setAutoStartup(true);
return rabbitAdmin;
}
}
- RabbitAdmin使用(希望能耐心看下代码里的注释,非常详细)
package com.example.demo1;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
/**
* @author:luzaichun
* @Date:2020/12/30
* @Time:21:28
**/
@SpringBootTest
@RunWith(SpringRunner.class)
public class Test1 {
@Autowired
private RabbitAdmin rabbitAdmin;
@Test
public void test1() {
rabbitAdmin.declareExchange(new TopicExchange("test.topic.exchange", false, false, null));
rabbitAdmin.declareExchange(new DirectExchange("test.direct.exchange", false, false, null));
rabbitAdmin.declareExchange(new FanoutExchange("test.fanout.exchange", false, false, null));
rabbitAdmin.declareQueue(new Queue("topic.queue", false, false, false, null));
rabbitAdmin.declareQueue(new Queue("direct.queue", false, false, false, null));
rabbitAdmin.declareQueue(new Queue("fanout.queue", false, false, false, null));
//1将一个queue绑定到一个exchange
rabbitAdmin.declareBinding(new Binding(
"direct.queue",//目标:队列名
Binding.DestinationType.QUEUE,//绑定目标类型:队列
"test.direct.exchange",//交换机名称
"direct.key",//路由key
null));//扩展参数
//2.1将一个交换机绑定到另一个交换机(消息流转topic.exchange->fanout.exchange)
rabbitAdmin.declareBinding(new Binding(
"test.fanout.exchange",//目标:交换机名
Binding.DestinationType.EXCHANGE, //绑定目标类型:交换机
"test.topic.exchange", //发起绑定的交换机
"test", //路由key
null));
//2.2fanout.queue绑定到test.fanout.exchange
rabbitAdmin.declareBinding(new Binding("fanout.queue",//目标:fanout.queue
Binding.DestinationType.QUEUE,//绑定类型:队列
"test.fanout.exchange",//绑定到的exchange
"",//应为是fanout类型exchange所以不需要routingKey
null));
//发送消息
//正常的消息流转 从test.direct.exchange-》direct.queue
rabbitAdmin.getRabbitTemplate().convertAndSend("test.direct.exchange", "direct.key", "直连交换机消息111");
//消息先到test.topic.exchange-》test.fanout.exchange-》fanout.queue
rabbitAdmin.getRabbitTemplate().convertAndSend("test.topic.exchange", "test", "多级流转消息2222");
}
}
- 观察
-
direct.queue队列有一条消息,这条消息是从test.direct.exchange交换机路由过来的
-
fanout.queue队列这条消息,消息先到达test.topic.exchange-》test.fanout.exchange-》fanout.queue。从下面2张图能看出来test.fanout.exchange绑定到test.topic.exchange,然后fanout.queue绑定到test.fanout.exchange
-
topic.queue队列没有消息是正常的,因为我们没有给他绑定exchange
补充:
1.另外补充一种绑定队列交换机的方式,通过在rabbitAdmin.declareBinding时候,通过BindingBuilder来声明exchange和queue并和设置routingKey
//3.另一种方式,在绑定的时候去声明队列和交换机以及路由key
rabbitAdmin.declareBinding(
BindingBuilder
.bind(new Queue("test001.queue", false, false, false))//需要绑定队列
.to(new TopicExchange("test001.exchange", false, false))//绑定到那个交换机
.with("topic.test001"));//路由key```
2.rabbitAdmin除了声明exchange queue binding 发送消息等api还有如:
- purgeQueue():清空队列
- deleteQueue():删除队列
- deleteExchange():删除交换机
- removeBinding():删除绑定
等等常用Api
更多推荐
所有评论(0)