目录
一、概念
二、临时队列
三、绑定
四、Fanout(扇出交换机)
(一)介绍
(二)实战
五、Direct(直接交换机)
(一)介绍
(二)实战
六、Topic(主题交换机)
(一)介绍
(二)案例
(三)实战
七、Header(头部交换机)
一、概念
交换机的类型:直接(direct), 主题(topic) ,标题(headers) , 扇出(fanout)
无名 exchange
channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes() );
二、临时队列
String queueName = channel.queueDeclare().getQueue();
三、绑定
四、Fanout(扇出交换机)
(一)介绍
(二)实战
我们声明一个logs交换机以及创建两个临时队列绑定在交换机上,routeKey为空字符串,当EmitLog生产者向logs交换机发消息时两个队列都接收得到。这个即为扇出交换机。
ReceiveLogs01 将接收到的消息打印在控制台
public class ReceiveLogs01 {
private final static String EXCHANGE_NAME = "logs";
public static void main(String[] args) throws IOException {
Channel channel = RabbitMqUtils.getChannel();
/*
* 生成一个临时的队列,队列名字随机
*/
String queueName = channel.queueDeclare().getQueue();
// 将临时队列绑定在我们定义的交换机上,路由Key为空字符串
channel.queueBind(queueName, EXCHANGE_NAME, "");
// 接收消息回调函数
DeliverCallback deliverCallback = (consumerTag, message) -> {
System.out.println("ReceiveLogs01控制台打印接收到的消息:" + new String(message.getBody()));
};
channel.basicConsume(queueName, true, deliverCallback, (consumerTag -> {}));
}
}
ReceiveLogs02 将接收到的消息打印在控制台
public class ReceiveLogs02 {
private final static String EXCHANGE_NAME = "logs";
public static void main(String[] args) throws IOException {
Channel channel = RabbitMqUtils.getChannel();
/*
* 生成一个临时的队列,队列名字随机
*/
String queueName = channel.queueDeclare().getQueue();
// 将临时队列绑定在我们定义的交换机上,路由Key为空字符串
channel.queueBind(queueName, EXCHANGE_NAME, "");
DeliverCallback deliverCallback = (consumerTag, message) -> {
System.out.println("ReceiveLogs02控制台打印接收到的消息:" + new String(message.getBody()));
};
channel.basicConsume(queueName, true, deliverCallback, (consumerTag -> {}));
}
}
EmitLog 发送消息给两个消费者接收
public class EmitLog {
private final static String EXCHANGE_NAME = "logs";
public static void main(String[] args) throws IOException {
// 获取信道
Channel channel = RabbitMqUtils.getChannel();
// 交换机声明,在生产者声明消费者方就不用声明了
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
// 控制台输入内容充当消息发送
Scanner scanner = new Scanner(System.in);
while(scanner.hasNext()) {
String message = scanner.next();
channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
System.out.println("生产者发出消息");
}
}
}
总结:
当交换机内的队列routeKey为空字符串时,此时向该交换机发送消息且routeKey也为空字符串的话那么所有的队列都接收得到消息,即为扇出交换机。
五、Direct(直接交换机)
(一)介绍
当我们想对某一个队列单独发送消息,可以根据设置不同的routeKey来达到这种效果,这就是直接交换机。
多重绑定无非就是绑定的时候多出一条queueBind
channel.queueBind(queue_name, EXCHANGE_NAME, "info");
channel.queueBind(queue_name, EXCHANGE_NAME, "waring");
(二)实战
此时队列的声明需要放在消费者方,因为每个消费者代表不同的队列,对于交换机的声明可以放在生产者
消费者01
public class ReceiveLogsDirect01 {
private static final String EXCHANGE_NAME = "direct_logs";
public static void main(String[] args) throws IOException {
Channel channel = RabbitMqUtils.getChannel();
String queue_name = "console";
channel.queueDeclare(queue_name, false, false, false, null);
channel.queueBind(queue_name, EXCHANGE_NAME, "info");
channel.queueBind(queue_name, EXCHANGE_NAME, "waring");
DeliverCallback deliverCallback = (consumerTag, message) -> {
System.out.println("ReceiveLogsDirect01收到消息" + new String(message.getBody()));
};
channel.basicConsume(queue_name, deliverCallback, (consumerTag -> {}));
}
}
消费者02
public class ReceiveLogsDirect02 {
private static final String EXCHANGE_NAME = "direct_logs";
public static void main(String[] args) throws IOException {
Channel channel = RabbitMqUtils.getChannel();
String queue_name = "disk";
channel.queueDeclare(queue_name, false, false, false, null);
channel.queueBind(queue_name, EXCHANGE_NAME, "error");
DeliverCallback deliverCallback = (consumerTag, message) -> {
System.out.println("ReceiveLogsDirect02收到消息" + new String(message.getBody()));
};
channel.basicConsume(queue_name, deliverCallback, (consumerTag -> {}));
}
}
生产者
public class EmitLogDirect {
private final static String EXCHANGE_NAME = "direct_logs";
public static void main(String[] args) throws IOException {
Channel channel = RabbitMqUtils.getChannel();
/*
* 生产者并不知道消息会发向哪个队列,他只是指明找哪个交换机中的路由key
* 由交换机去寻找队列传递消息
*/
Scanner sc = new Scanner(System.in);
while(sc.hasNext()) {
String message = sc.next();
channel.basicPublish(EXCHANGE_NAME, "info", null, message.getBytes());
}
}
}
总结:
我们可以通过指定routeKey来向对应的队列单独发消息,即为直接交换机
六、Topic(主题交换机)
(一)介绍
- *(星号)可以代替一个单词
- #(井号)可以替代零个或多个单词
(二)案例
队列Q1的routeKey为“*.orange.*”,Q2的routeKey有两个,分别为“*.*.rabbit”和“lazy.#”
- 当一个队列绑定键是#,那么这个队列将接收所有数据,就有点像 fanout 了
- 如果队列绑定键当中没有#和*出现,那么该队列绑定类型就是 direct 了
(三)实战
消费者01
public class ReceiveLogsTopic01 {
public static final String EXCHANGE_NAME = "topic_logs";
public static void main(String[] args) throws IOException {
Channel channel = RabbitMqUtils.getChannel();
String queueName = "Q1";
channel.queueDeclare(queueName, false, false, false, null);
channel.queueBind(queueName, EXCHANGE_NAME, "*.orange.*");
System.out.println("ReceiveLogsTopic01等待接收消息...");
DeliverCallback deliverCallback = (consumeTag, message) -> {
System.out.println("接收队列:" + queueName + "绑定键:" + message.getEnvelope().getRoutingKey());
System.out.println("接收到消息:" + new String(message.getBody()));
};
channel.basicConsume(queueName, true, deliverCallback, (consumerTag -> {}));
}
}
消费者02
public class ReceiveLogsTopic02 {
public static final String EXCHANGE_NAME = "topic_logs";
public static void main(String[] args) throws IOException {
Channel channel = RabbitMqUtils.getChannel();
String queueName = "Q2";
channel.queueDeclare(queueName, false, false, false, null);
channel.queueBind(queueName, EXCHANGE_NAME, "*.*.rabbit");
channel.queueBind(queueName, EXCHANGE_NAME, "lazy.#");
System.out.println("ReceiveLogsTopic02等待接收消息...");
DeliverCallback deliverCallback = (consumeTag, message) -> {
System.out.println("接收队列:" + queueName + "绑定键:" + message.getEnvelope().getRoutingKey());
System.out.println("接收到消息:" + new String(message.getBody()));
};
channel.basicConsume(queueName, true, deliverCallback, (consumerTag -> {}));
}
}
生产者
public class EmitLogTopic {
public static final String EXCHANGE_NAME = "topic_logs";
public static void main(String[] args) throws IOException {
Channel channel = RabbitMqUtils.getChannel();
channel.exchangeDeclare(EXCHANGE_NAME, "topic");
Map<String, String> bindingKeyMap = new HashMap<>();
bindingKeyMap.put("quick.orange.rabbit","被队列 Q1Q2 接收到");
bindingKeyMap.put("lazy.orange.elephant","被队列 Q1Q2 接收到");
bindingKeyMap.put("quick.orange.fox","被队列 Q1 接收到");
bindingKeyMap.put("lazy.brown.fox","被队列 Q2 接收到");
bindingKeyMap.put("lazy.pink.rabbit","虽然满足两个绑定但只被队列 Q2 接收一次");
bindingKeyMap.put("quick.brown.fox","不匹配任何绑定不会被任何队列接收到会被丢弃");
bindingKeyMap.put("quick.orange.male.rabbit","是四个单词不匹配任何绑定会被丢弃");
bindingKeyMap.put("lazy.orange.male.rabbit","是四个单词但匹配 Q2");
for(Map.Entry<String, String> bindingKeyMapEntry : bindingKeyMap.entrySet()) {
String bindingKey = bindingKeyMapEntry.getKey();
String message = bindingKeyMapEntry.getValue();
channel.basicPublish(EXCHANGE_NAME, bindingKey, null, message.getBytes());
System.out.println("生产者发出消息:" + message);
}
}
}
结果
七、Header(头部交换机)
由于header交换机不常用,这里简单介绍