Vhosts 虚拟机
虚拟机可以理解为RabbitMQ的独立划分空间（有点类似mini版的RabbitMQ）,RabbitMQ支持多个虚拟机,虚拟机里面不能有同名的交换机。
Exchange交换机
交换机可以RabbitMQ的消息处理中心（类似快递分拣中心）负责消息分类管理
交换机有4种消息投递模式
-
1.Direct模式（根据路由key 直连模式，完全匹配模式）需要声明队列
1.代码示例
/**
生产端
**/
private static final String QUEUE_NAME = "test_simple_queue";
private static final String EXCHANGE_DIRECT = "test_exchange_direct";
/**
* direct 模式
* 只有队列名称完全比配上才能送达
*/
public static void exchange_direct(String msg){
try {
//获取连接
Connection connection = ConnectionUtil.getConnection();
//从连接中获取一个通道
Channel channel = connection.createChannel();
//创建交换机
channel.exchangeDeclare(EXCHANGE_DIRECT, BuiltinExchangeType.DIRECT);
String message = msg;
//发送消息
channel.basicPublish(EXCHANGE_DIRECT, QUEUE_NAME, null, message.getBytes("utf-8"));
System.out.println("[发送]：" + message);
channel.close();
connection.close();
}
catch (Exception e) {
e.printStackTrace();
}
}
/**
* 消费端
*/
public class Receive {
private static final String QUEUE_NAME = "test_simple_queue";
public static void main(String[] args) {
try {
//获取连接
Connection connection = ConnectionUtil.getConnection();
//从连接中获取一个通道
Channel channel = connection.createChannel();
//声明队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
//定义消费者
DefaultConsumer consumer = new DefaultConsumer(channel) {
//当消息到达时执行回调方法
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
byte[] body) throws IOException {
String message = new String(body, "utf-8");
System.out.println("[消费]：" + message);
}
};
//监听队列
channel.basicConsume(QUEUE_NAME, true, consumer);
}
catch (Exception e) {
e.printStackTrace();
}
}
}
-
2.Topic模式（根据路由key 模糊匹配模式 ）需要声明队列
1.代码示例
private static final String EXCHANGE_TOPIC = "test_exchange_topic";
public static final String ROUTING_KEY = "*.topic";
/**
* topic 模式
* 根据路由key模糊匹配功
* *仅代表一个单词 *.topic 能匹配 a.topic,b.topic 等
* #代表任意个单词 #.topic 能匹配 a.b.c.topic 等
*/
public static void exchange_topic(String msg){
try {
//获取连接
Connection connection = ConnectionUtil.getConnection();
//从连接中获取一个通道
Channel channel = connection.createChannel();
//创建交换机
channel.exchangeDeclare(EXCHANGE_TOPIC, BuiltinExchangeType.TOPIC);
//声明队列
String message = msg;
//发送消息
channel.basicPublish(EXCHANGE_TOPIC, ROUTING_KEY, null, message.getBytes("utf-8"));
System.out.println("[发送]：" + message);
channel.close();
connection.close();
}
catch (Exception e) {
e.printStackTrace();
}
}
/**
* topic 模式 消费端
*/
private static final String QUEUE_NAME2 = "test.topic";
private static final String EXCHANGE_TOPIC = "test_exchange_topic";
//路由
public static final String ROUTING_KEY = "*.topic";
try {
//获取连接
Connection connection = ConnectionUtil.getConnection();
//从连接中获取一个通道
Channel channel = connection.createChannel();
//声明队列
channel.queueDeclare(QUEUE_NAME2, false, false, false, null);
//绑定交换机
channel.queueBind(QUEUE_NAME2, EXCHANGE_TOPIC, "");
//定义消费者
DefaultConsumer consumer = new DefaultConsumer(channel) {
//当消息到达时执行回调方法
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
byte[] body) throws IOException {
String message = new String(body, "utf-8");
System.out.println("[消费2]：" + message);
}
};
//监听队列
channel.basicConsume(QUEUE_NAME2, true, consumer);
}
catch (Exception e) {
e.printStackTrace();
}
}
-
3.Fanout模式 (广播模式,绑定该交换机的列队都能接受) 无需声明队列
1.代码示例
private static final String EXCHANGE_FANOUT = "test_exchange_fanout";
/**
* fanout 模式
* 绑定了同一交换机内的所有队列都能接收
*/
public static void exchange_fanout(String msg){
try {
//获取连接
Connection connection = ConnectionUtil.getConnection();
//从连接中获取一个通道
Channel channel = connection.createChannel();
//创建交换机
channel.exchangeDeclare(EXCHANGE_FANOUT, BuiltinExchangeType.FANOUT);
//声明队列
String message = msg;
//发送消息
channel.basicPublish(EXCHANGE_FANOUT, "", null, message.getBytes("utf-8"));
System.out.println("[发送]：" + message);
channel.close();
connection.close();
}
catch (Exception e) {
e.printStackTrace();
}
}
/**
* 消费端
*
*/
private static final String EXCHANGE_FANOUT = "test_exchange_fanout";
try {
//获取连接
Connection connection = ConnectionUtil.getConnection();
//从连接中获取一个通道
Channel channel = connection.createChannel();
//声明队列
channel.queueDeclare("", false, false, false, null);
//绑定交换机
channel.queueBind("", EXCHANGE_FANOUT, "");
//定义消费者
DefaultConsumer consumer = new DefaultConsumer(channel) {
//当消息到达时执行回调方法
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
byte[] body) throws IOException {
String message = new String(body, "utf-8");
System.out.println("[消费2]：" + message);
}
};
//监听队列
channel.basicConsume(QUEUE_NAME3, true, consumer);
}
catch (Exception e) {
e.printStackTrace();
}
-
4.Headers 模式（不处理路由键。而是根据发送的消息内容中的headers属性进行匹配） 无需声明队列
1.代码示例
/**
* header模式
* 根据消息头的建 值匹配消息
* AMQP.BasicProperties
* 注意点 x-match 有两个属性, any（满足一条即可）,all（必须满足所有）
*/
public static void exchange_header(String msg){
try {
//获取连接
Connection connection = ConnectionUtil.getConnection();
//从连接中获取一个通道
Channel channel = connection.createChannel();
//创建交换机
channel.exchangeDeclare(EXCHANGE_HEADER, BuiltinExchangeType.HEADERS);
//声明队列
String message = msg;
//设置消息头
Map<String,Object> map = new HashMap<>();
map.put("name","fzh1");
map.put("x-match","any");
// map.put("x-match","all");
AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().headers(map).build();
//发送消息
channel.basicPublish(EXCHANGE_HEADER, "", properties, message.getBytes("utf-8"));
System.out.println("[发送]：" + message);
channel.close();
connection.close();
}
catch (Exception e) {
e.printStackTrace();
}
/**
* 消费端
*/
private static final String QUEUE_NAME = "test.header";
private static final String EXCHANGE_HEADER = "test_exchange_header";
try {
//获取连接
Connection connection = ConnectionUtil.getConnection();
//从连接中获取一个通道
Channel channel = connection.createChannel();
Map<String,Object> map = new HashMap<>();
map.put("name","fzh1");
map.put("x-match","any");
//绑定交换机
channel.exchangeDeclare(EXCHANGE_HEADER, BuiltinExchangeType.HEADERS);
//声明队列
channel.queueDeclare("", false, false, false, null);
//绑定交换机
channel.queueBind("",EXCHANGE_HEADER,"",map);
//定义消费者
DefaultConsumer consumer = new DefaultConsumer(channel) {
//当消息到达时执行回调方法
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
byte[] body) throws IOException {
String message = new String(body, "utf-8");
System.out.println("[消费2]：" + message+"___"+properties.getHeaders().toString());
}
};
//监听队列
channel.basicConsume(QUEUE_NAME, true, consumer);
}
catch (Exception e) {
e.printStackTrace();
}
路由
交换机跟队列之间的链接点 ，再 Direct模式 跟topic模式下 必须绑定
队列
具体消息存储的地方
生产者
发送消息端
消费者
接受消息消费
