[TOC]
交换机
- 用于接收客户端投递过来的消息
- 分发消息到指定的队列
常用类型
- 目前共四种类型：direct、fanout、topic、headers(headers匹配AMQP消息的header而不是路由键(Routing-key)，此外headers交换器和direct交换器完全一致，但是性能差了很多，目前几乎用不到了。所以直接看另外三种类型)。
direct
image.png
直连方式，如果routingkey和bindingkey一致，交换机会把消息发往该队列。
消费端例子
package com.coder.wen.exchange.direct;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* direct 类型
* 消费者
*
* @author wen
*/
public class Consumer1 {
public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("192.168.205.10");
factory.setPort(5672);
factory.setVirtualHost("/");
/*1. 从工厂创建连接**/
Connection connection = factory.newConnection();
/**
* 2. 从connection拿到channel
*/
Channel channel = connection.createChannel();
/**
* 声明交换机
* 参数1：交换机名称
* 参数2：交换机类型
* 参数3 是否持久化（是否存储磁盘）
* 参数4：当交换机不再使用时候时候自动删除
* 参数5：扩展参数
*/
String exhangeType = "direct";
String exchangeName = "exchange.test.direct";
channel.exchangeDeclare(exchangeName, exhangeType, true, false, null);
/**
* 声明队列
* 参数1：队列名称
* @参数2： 是否持久化
* @参数3： 是否独占，如果是的话，当一个连接使用了这个队列，其他连接就不能使用
* @参数4：队列不再使用的时候是否自动删除
* 参数5：扩展参数
*/
String queueName = "direct.queue";
channel.queueDeclare(queueName, true, false, false, null);
/**
* 交换机和队列绑定
* 参数1：队列名称
* @参数2：交换机名称
* 参数3：路由键
* 参数4：扩展参数
*/
String routingkeyName = "direct.key";
channel.queueBind(queueName, exchangeName, routingkeyName, null);
QueueingConsumer consumer = new QueueingConsumer(channel);
// 参数：队列名称、是否自动ACK、Consumer
channel.basicConsume(queueName, true, consumer);
// 循环获取消息
while (true) {
// 获取消息，如果没有消息，这一步将会一直阻塞
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
String msg = new String(delivery.getBody());
System.out.println("收到消息：" + msg);
}
}
}
- 生产端
package com.coder.wen.exchange.direct;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* 生产者
* direct 演示
*
* @author wen
*/
public class Producer {
public static void main(String[] args) throws IOException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("192.168.205.10");
factory.setPort(5672);
factory.setVirtualHost("/");
/*1. 从工厂创建连接**/
Connection connection = factory.newConnection();
/**
* 2. 从connection拿到channel
*/
Channel channel = connection.createChannel();
/**
* 声明交换机 和路由 名称
*/
String exchange = "exchange.test.direct";
String routingkey = "direct.key";
/**
* 发送消息
*/
channel.basicPublish(exchange, routingkey, null, "hello world".getBytes());
/**
* 关闭连接
*/
channel.close();
connection.close();
}
}
-
运行结果（注意，首先运行消费端，因为需要创建交换机、队列、以及绑定关系）
image.png
topic
image.png
路由键和某个模式进行匹配，此时队列需要绑定到一个模式上。它同样也会识别两个通配符："#"和" * "。#匹配0个或多个单词，" * "匹配不多不少一个单词。
消费端例子
package com.coder.wen.exchange.topic;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.QueueingConsumer.Delivery;
public class Consumer1 {
public static void main(String[] args) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("192.168.205.10");
factory.setPort(5672);
factory.setVirtualHost("/");
factory.setAutomaticRecoveryEnabled(true);
factory.setNetworkRecoveryInterval(3000);
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
//4 声明
String exchangeName = "topic.exchange";
String exchangeType = "topic";
String queueName = "topic.queue";
String routingKey = "user.*";
// String routingKey = "user.#";
channel.exchangeDeclare(exchangeName, exchangeType, true, false, false, null);
channel.queueDeclare(queueName, false, false, false, null);
channel.queueBind(queueName, exchangeName, routingKey);
//durable 是否持久化消息
QueueingConsumer consumer = new QueueingConsumer(channel);
// 参数：队列名称、是否自动ACK、Consumer
channel.basicConsume(queueName, true, consumer);
// 循环获取消息
while (true) {
// 获取消息，如果没有消息，这一步将会一直阻塞
Delivery delivery = consumer.nextDelivery();
String msg = new String(delivery.getBody());
System.out.println("收到消息：" + msg + ", RoutingKey: " + delivery.getEnvelope().getRoutingKey());
}
}
}
- 生产端例子
package com.coder.wen.exchange.topic;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class Producer {
public static void main(String[] args) throws Exception {
//1 创建ConnectionFactory
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("192.168.205.10");
factory.setPort(5672);
factory.setVirtualHost("/");
//2 创建Connection
Connection connection = factory.newConnection();
//3 创建Channel
Channel channel = connection.createChannel();
//4 声明
String exchangeName = "topic.exchange";
String routingKey1 = "user.save";
String routingKey2 = "user.update";
String routingKey3 = "user.delete.abc";
//5 发送
String msg = "Hello World Topic Exchange Message ...";
channel.basicPublish(exchangeName, routingKey1, null, msg.getBytes());
channel.basicPublish(exchangeName, routingKey2, null, msg.getBytes());
channel.basicPublish(exchangeName, routingKey3, null, msg.getBytes());
channel.close();
connection.close();
}
}
- 结果
生产端发送了三条消息，消费端只能接收到routingKey为user.save、user.update的这两条消息，因为bindingKey = "user.* "，“ * ”只能匹配到一个字符。如果此时换成“#”，那这三条消息都能接收到。
fanout
image.png
英文字面意思：扇形。fanout属于广播模式，
意思就是消息不走任何的路由规则，只有队列和交换机有绑定关系就能收到消息。消费端例子
package com.coder.wen.exchange.fanout;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.QueueingConsumer.Delivery;
public class Consumer1 {
public static void main(String[] args) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("192.168.205.10");
factory.setPort(5672);
factory.setVirtualHost("/");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
//4 声明
String exchangeName = "fanout.exchange";
String exchangeType = "fanout";
String queueName = "fanout.queue";
String routingKey = ""; // 不设置路由键
channel.exchangeDeclare(exchangeName, exchangeType, true, false, false, null);
channel.queueDeclare(queueName, false, false, false, null);
channel.queueBind(queueName, exchangeName, routingKey);
//durable 是否持久化消息
QueueingConsumer consumer = new QueueingConsumer(channel);
//参数：队列名称、是否自动ACK、Consumer
channel.basicConsume(queueName, true, consumer);
//循环获取消息
while (true) {
//获取消息，如果没有消息，这一步将会一直阻塞
Delivery delivery = consumer.nextDelivery();
String msg = new String(delivery.getBody());
System.out.println("收到消息：" + msg);
}
}
}
- 再声明一个消费者
package com.coder.wen.exchange.fanout;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.QueueingConsumer.Delivery;
public class Consumer2 {
public static void main(String[] args) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("192.168.205.10");
factory.setPort(5672);
factory.setVirtualHost("/");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
//4 声明
String exchangeName = "fanout.exchange";
String exchangeType = "fanout";
String queueName = "fanout.queue";
String routingKey = ""; // 不设置路由键
channel.exchangeDeclare(exchangeName, exchangeType, true, false, false, null);
channel.queueDeclare(queueName, false, false, false, null);
channel.queueBind(queueName, exchangeName, routingKey);
//durable 是否持久化消息
QueueingConsumer consumer = new QueueingConsumer(channel);
//参数：队列名称、是否自动ACK、Consumer
channel.basicConsume(queueName, true, consumer);
//循环获取消息
while (true) {
//获取消息，如果没有消息，这一步将会一直阻塞
Delivery delivery = consumer.nextDelivery();
String msg = new String(delivery.getBody());
System.out.println("收到消息：" + msg);
}
}
}
- 生产端例子
package com.coder.wen.exchange.fanout;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class Producer {
public static void main(String[] args) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("192.168.205.10");
factory.setPort(5672);
factory.setVirtualHost("/");
//2 创建Connection
Connection connection = factory.newConnection();
//3 创建Channel
Channel channel = connection.createChannel();
//4 声明
String exchangeName = "fanout.exchange";
//5 发送
for (int i = 0; i < 10; i++) {
String msg = "Hello World FANOUT Exchange Message ...";
channel.basicPublish(exchangeName, "", null, msg.getBytes());
}
channel.close();
connection.close();
}
}
-
结果
image.png
image.png
生产者发送了5条消息，消费者1、2 分别都收到了5条消息。
- 最后，交换机的属性整理一下
- Name: 交换机名称
- Type: 交换机类型，direct、topic、 fanout、 headers
- Durability: 是否需要持久化
- Auto Delete: 当最后一个绑定到Exchange上的队列删除后，自动删除该Exchange
- Internal: 当前Exchange是否用于RabbitMQ内部使用，默认为False
- Arguments: 扩展参数，用于扩展AMQP协议定制化使用
- 个人工作之余做的笔记，会一直更新，如果有错还请指出，共同进步。
