Vhosts 虛擬機
虛擬機可以理解為RabbitMQ的獨立劃分空間（有點類似mini版的RabbitMQ）,RabbitMQ支持多個虛擬機,虛擬機里面不能有同名的交換機于样。
Exchange交換機
交換機可以RabbitMQ的消息處理中心（類似快遞分揀中心）負(fù)責(zé)消息分類管理
交換機有4種消息投遞模式
-
1.Direct模式（根據(jù)路由key 直連模式，完全匹配模式）需要聲明隊列
1.代碼示例
/**
生產(chǎn)端
**/
private static final String QUEUE_NAME = "test_simple_queue";
private static final String EXCHANGE_DIRECT = "test_exchange_direct";
/**
* direct 模式
* 只有隊列名稱完全比配上才能送達(dá)
*/
public static void exchange_direct(String msg){
try {
//獲取連接
Connection connection = ConnectionUtil.getConnection();
//從連接中獲取一個通道
Channel channel = connection.createChannel();
//創(chuàng)建交換機
channel.exchangeDeclare(EXCHANGE_DIRECT, BuiltinExchangeType.DIRECT);
String message = msg;
//發(fā)送消息
channel.basicPublish(EXCHANGE_DIRECT, QUEUE_NAME, null, message.getBytes("utf-8"));
System.out.println("[發(fā)送]：" + 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) {
//當(dāng)消息到達(dá)時執(zhí)行回調(diào)方法
@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);
}
};
//監(jiān)聽隊列
channel.basicConsume(QUEUE_NAME, true, consumer);
}
catch (Exception e) {
e.printStackTrace();
}
}
}
-
2.Topic模式（根據(jù)路由key 模糊匹配模式 ）需要聲明隊列
1.代碼示例
private static final String EXCHANGE_TOPIC = "test_exchange_topic";
public static final String ROUTING_KEY = "*.topic";
/**
* topic 模式
* 根據(jù)路由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();
//創(chuàng)建交換機
channel.exchangeDeclare(EXCHANGE_TOPIC, BuiltinExchangeType.TOPIC);
//聲明隊列
String message = msg;
//發(fā)送消息
channel.basicPublish(EXCHANGE_TOPIC, ROUTING_KEY, null, message.getBytes("utf-8"));
System.out.println("[發(fā)送]：" + 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) {
//當(dāng)消息到達(dá)時執(zhí)行回調(diào)方法
@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);
}
};
//監(jiān)聽隊列
channel.basicConsume(QUEUE_NAME2, true, consumer);
}
catch (Exception e) {
e.printStackTrace();
}
}
-
3.Fanout模式 (廣播模式,綁定該交換機的列隊都能接受) 無需聲明隊列
1.代碼示例
private static final String EXCHANGE_FANOUT = "test_exchange_fanout";
/**
* fanout 模式
* 綁定了同一交換機內(nèi)的所有隊列都能接收
*/
public static void exchange_fanout(String msg){
try {
//獲取連接
Connection connection = ConnectionUtil.getConnection();
//從連接中獲取一個通道
Channel channel = connection.createChannel();
//創(chuàng)建交換機
channel.exchangeDeclare(EXCHANGE_FANOUT, BuiltinExchangeType.FANOUT);
//聲明隊列
String message = msg;
//發(fā)送消息
channel.basicPublish(EXCHANGE_FANOUT, "", null, message.getBytes("utf-8"));
System.out.println("[發(fā)送]：" + 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) {
//當(dāng)消息到達(dá)時執(zhí)行回調(diào)方法
@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);
}
};
//監(jiān)聽隊列
channel.basicConsume(QUEUE_NAME3, true, consumer);
}
catch (Exception e) {
e.printStackTrace();
}
-
4.Headers 模式（不處理路由鍵。而是根據(jù)發(fā)送的消息內(nèi)容中的headers屬性進(jìn)行匹配） 無需聲明隊列
1.代碼示例
/**
* header模式
* 根據(jù)消息頭的建 值匹配消息
* AMQP.BasicProperties
* 注意點 x-match 有兩個屬性, any（滿足一條即可）,all（必須滿足所有）
*/
public static void exchange_header(String msg){
try {
//獲取連接
Connection connection = ConnectionUtil.getConnection();
//從連接中獲取一個通道
Channel channel = connection.createChannel();
//創(chuàng)建交換機
channel.exchangeDeclare(EXCHANGE_HEADER, BuiltinExchangeType.HEADERS);
//聲明隊列
String message = msg;
//設(shè)置消息頭
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();
//發(fā)送消息
channel.basicPublish(EXCHANGE_HEADER, "", properties, message.getBytes("utf-8"));
System.out.println("[發(fā)送]：" + 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) {
//當(dāng)消息到達(dá)時執(zhí)行回調(diào)方法
@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());
}
};
//監(jiān)聽隊列
channel.basicConsume(QUEUE_NAME, true, consumer);
}
catch (Exception e) {
e.printStackTrace();
}
路由
交換機跟隊列之間的鏈接點 ，再 Direct模式 跟topic模式下 必須綁定
隊列
具體消息存儲的地方
生產(chǎn)者
發(fā)送消息端
消費者
接受消息消費
