RabbitMQ系列—Java操作之Publish/Subscribe、路由模式、Topics主题模式

亦凉 2022-05-08 10:24 370阅读 0赞

### Exchange交换机 ###

在后面介绍的Publish/Subscribe、路由模式、主题模式都需要用到交换机，在引入了交换机组件后，重新归纳一下RabbitMQ的各组件功能。

*  生产者：发送消息
 *  交换机：将收到的消息根据路由规则路由到特定队列
 *  队列：用于存储消息
 *  消费者：收到消息并消费

**交换机的类型**

交换机主要包括如下4种类型：

1.  Direct exchange（直连交换机）
2.  Fanout exchange（扇型交换机）
3.  Topic exchange（主题交换机）
4.  Headers exchange（头交换机）

另外RabbitMQ默认定义一些交换机：

*  默认交换机
 *  amq.\* exchanges

还有一类特殊的交换机：Dead Letter Exchange（死信交换机）

**Direct exchange（直连交换机）**：是根据消息携带的路由键（routing key）将消息投递给对应队列的，步骤如下：

*  将一个队列绑定到某个交换机上，同时赋予该绑定一个路由键（routing key）
 *  当一个携带着路由值为R的消息被发送给直连交换机时，交换机会把它路由给绑定值同样为R的队列。

**Fanout exchange（扇型交换机）**：扇型交换机（funout exchange）将消息路由给绑定到它身上的所有队列。不同于直连交换机，路由键在此类型上不启任务作用。如果N个队列绑定到某个扇型交换机上，当有消息发送给此扇型交换机时，交换机会将消息的发送给这所有的N个队列。

**Topic exchange（主题交换机）**：主题交换机（topic exchanges）中，队列通过路由键绑定到交换机上，然后，交换机根据消息里的路由键，将消息路由给一个或多个绑定队列。

扇型交换机和主题交换机异同：

*  对于扇型交换机路由键是没有意义的，只要有消息，它都发送到它绑定的所有队列上。
 *  对于主题交换机，路由规则由路由键决定，只有满足路由键的规则，消息才可以路由到对应的队列上。

**Headers exchange（头交换机）**：类似主题交换机，但是头交换机使用多个消息属性来代替路由键建立路由规则。通过判断消息头的值能否与指定的绑定相匹配来确立路由规则。 此交换机有个重要参数：”x-match”：

*  当”x-match”为“any”时，消息头的任意一个值被匹配就可以满足条件
 *  当”x-match”设置为“all”的时候，就需要消息头的所有值都匹配成功

**默认交换机（default exchange）**：实际上它是一个由RabbitMQ预先声明好的名字为空字符串的直连交换机（direct exchange）。它有一个特殊的属性使得它对于简单应用特别有用处：那就是每个新建队列（queue）都会自动绑定到默认交换机上，绑定的路由键（routing key）名称与队列名称相同。

如：当你声明了一个名为”hello”的队列，RabbitMQ会自动将其绑定到默认交换机上，绑定（binding）的路由键名称也是为”hello”。因此，当携带着名为”hello”的路由键的消息被发送到默认交换机的时候，此消息会被默认交换机路由至名为”hello”的队列中。即默认交换机看起来貌似能够直接将消息投递给队列，如同我们之前写的简单队列和工作队列，在channel.basicPublish(）方法中，第一个参数就是路由名称，它的值设置的是空字符串""，即匿名转发。。

**类似amq.\*的名称的交换机**

这些是RabbitMQ默认创建的交换机。这些队列名称被预留做RabbitMQ内部使用，不能被应用使用，否则抛出403 (ACCESS\_REFUSED)错误。

![70][]

**Dead Letter Exchange（死信交换机）：**在默认情况，如果消息在投递到交换机时，交换机发现此消息没有匹配的队列，则这个消息将被悄悄丢弃。为了解决这个问题，RabbitMQ中有一种交换机叫死信交换机。当消费者不能处理接收到的消息时，将这个消息重新发布到另外一个队列中，等待重试或者人工干预。这个过程中的exchange和queue就是所谓的"Dead Letter Exchange 和 Queue"。

### Publish/Subscribe ###

![70 1][]

从模型中看到，P代表生产者，X代表交换机，红色代表对列，C1和C2代表消费者。和之前的简单队列和工作队列不同的是Publish/Subscribe中新增了交换机的概念，生产者没有把消息直接发送到队列内，而是发送给交换机，队列和交换机绑定，每个消费者监听各自的队列，这样就实现一个消息被多个消费者消费。这就是Publish/Subscribe的工作原理。之前介绍RabbitMQ的应用场景时说到的注册后发送邮件和短信就可以采用这种模式来实现。

**生产者Sender**

public class Sender {
    	/**
    	 * 交换机名称
    	 */
    	private static final String EXCHANGE = "test_exchange_fanout";
    
    	public static void main(String[] args) {
    		Connection con = null;
    		Channel channel = null;
    		try {
    			// 获取连接
    			con = ConnectionUtils.getConnection();
    			// 从连接中创建通道
    			channel = con.createChannel();
    			// 声明交换机,采用fanout（分发）类型
    			channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.FANOUT,false,false,null);
    			// 消息
    			String msg = "你好 exchange fanout!";
    			// 发送消息
    			channel.basicPublish(EXCHANGE, "", null, msg.getBytes());
    			System.out.println("send success");
    		} catch (IOException e) {
    			e.printStackTrace();
    		} catch (TimeoutException e) {
    			e.printStackTrace();
    		} finally {
    			// 关闭连接
    			ConnectionUtils.close(channel, con);
    		}
    	}
    }

运行一下Sender，打印send success。登录Web控制台，查看Exchanges多了一个交换机。

![70 2][]

此时消息已经丢失了，因为交换机并没有存储消息的能力，rabbitmq中只有队列可以保存消息，此时还没有队列和交换机绑定，现在定义消费者声明队列并和交换机绑定。

**消费者Recver1**

public class Recver1 {
    	/**
    	 * 交换机名称
    	 */
    	private static final String EXCHANGE = "test_exchange_fanout";
    	private static final String QUEUE = "test_queue_email";
    
    	public static void main(String[] args) throws IOException, TimeoutException {
    		// 获取连接
    		Connection con = ConnectionUtils.getConnection();
    		// 从连接中创建通道
    		final Channel channel = con.createChannel();
    		// 声明队列
    		channel.queueDeclare(QUEUE, true, false, false, null);
    		// 绑定交换机
    		channel.queueBind(QUEUE, EXCHANGE, "");
    		// 保证一次只分发一条消息
    		channel.basicQos(1);
    		// 创建消费者
    		Consumer consumer = new DefaultConsumer(channel) {
    			// 获取消息
    			@Override
    			public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
    					throws IOException {
    				String msg = new String(body, "utf-8");
    				System.out.println("Recver1接收到消息——" + msg);
    				// 手动确认消息
    				channel.basicAck(envelope.getDeliveryTag(), false);
    			}
    
    		};
    		// 监听队列
    		channel.basicConsume(QUEUE, false, consumer);
    	}
    }

**消费者Recver2**

public class Recver2 {
    	/**
    	 * 交换机名称
    	 */
    	private static final String EXCHANGE = "test_exchange_fanout";
    	private static final String QUEUE = "test_queue_sms";
    
    	public static void main(String[] args) throws IOException, TimeoutException {
    		// 获取连接
    		Connection con = ConnectionUtils.getConnection();
    		// 从连接中创建通道
    		final Channel channel = con.createChannel();
    		// 声明队列
    		channel.queueDeclare(QUEUE, true, false, false, null);
    		// 绑定交换机
    		channel.queueBind(QUEUE, EXCHANGE, "");
    		// 保证一次只分发一条消息
    		channel.basicQos(1);
    		// 创建消费者
    		Consumer consumer = new DefaultConsumer(channel) {
    			// 获取消息
    			@Override
    			public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
    					throws IOException {
    				String msg = new String(body, "utf-8");
    				System.out.println("Recver2接收到消息——" + msg);
    				// 手动确认消息
    				channel.basicAck(envelope.getDeliveryTag(), false);
    			}
    
    		};
    		// 监听队列
    		channel.basicConsume(QUEUE, false, consumer);
    	}
    }

依次启动消费者Recver1、消费者Recver2、生产者Sender，控制台打印（运行消费者之前必须先保证交换机已存在，否则会报错）

![70 3][]

![70 4][]

此时再去Web控制台查看发现交换机绑定了两个队列

![70 5][]

## ##

channel.exchangeDeclare()方法的第二个参数，是交换机的类型，它有如下几种：

public enum BuiltinExchangeType {
    
        DIRECT("direct"), FANOUT("fanout"), TOPIC("topic"), HEADERS("headers");
    
       
    }

在Publish/Subscribe中使用的是fanout类型（不处理路由键），它并没有使用路由键，可以看到channel.basicPublish()中第二个参数即路由键写的是空字符串""，它会把消息发送到所有绑定到此交换机的队列中。

channel.exchangeDeclare()方法的参数解释：

*  String exchange：交换机名称。
 *  BuiltinExchangeType type：交换机类型的枚举类。
 *  boolean durable：是否可持久化，如果为true，重启rabbitmq后交换机不会丢失。
 *  boolean autoDelete：是否自动删除。
 *  Map<String, Object> arguments：其他参数。

### 路由模式 ###

![70 6][]

路由模式将交换机定义为direct类型，每个队列都绑定到交换机上，并且都有自己的路由键（routing key），一个队列可以有多个路由键。同时发送消息时也有一个路由键（routing key），交换机会把消息转发给路由键匹配的队列上。

比如：Q1队列有一个error路由键，Q2有三个路由键分别是error、info和warning。如果交换机的消息的路由键是error则转发到Q1和Q2，如果是info或者warning则就转发给Q2。

**生产者Sender**

public class Sender {
    
    	private static final String EXCHANGE = "test_exchange_direct";
    	
    	public static void main(String[] args) {
    		Connection con = null;
    		Channel channel = null;
    		try {
    			// 获取连接
    			con = ConnectionUtils.getConnection();
    			// 从连接中创建通道
    			channel = con.createChannel();
    			// 声明交换机,采用direct类型
    			channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.DIRECT,false,false,null);
    			// 消息
    			String msg = "你好 exchange direct!";
    			// 发送消息,指定路由键
    			channel.basicPublish(EXCHANGE, "info", null, msg.getBytes());
    			System.out.println("send success");
    		} catch (IOException e) {
    			e.printStackTrace();
    		} catch (TimeoutException e) {
    			e.printStackTrace();
    		} finally {
    			// 关闭连接
    			ConnectionUtils.close(channel, con);
    		}
    	}
    }

**消费者Recver1**，指定路由键为error

public class Recver1 {
    	private static final String EXCHANGE = "test_exchange_direct";
    	private static final String QUEUE = "test_queue_direct_1";
    
    	public static void main(String[] args) throws IOException, TimeoutException {
    		// 获取连接
    		Connection con = ConnectionUtils.getConnection();
    		// 从连接中创建通道
    		final Channel channel = con.createChannel();
    		// 声明队列
    		channel.queueDeclare(QUEUE, true, false, false, null);
    		// 绑定交换机,指定路由键为error
    		channel.queueBind(QUEUE, EXCHANGE, "error");
    		// 保证一次只分发一条消息
    		channel.basicQos(1);
    		// 创建消费者
    		Consumer consumer = new DefaultConsumer(channel) {
    			// 获取消息
    			@Override
    			public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
    					throws IOException {
    				String msg = new String(body, "utf-8");
    				System.out.println("Recver1接收到消息——" + msg);
    				// 手动确认消息
    				channel.basicAck(envelope.getDeliveryTag(), false);
    			}
    
    		};
    		// 监听队列
    		channel.basicConsume(QUEUE, false, consumer);
    	}
    }

**消费者Recver2**，指定路由键为error、info、warning

public class Recver2 {
    	private static final String EXCHANGE = "test_exchange_direct";
    	private static final String QUEUE = "test_queue_direct_2";
    
    	public static void main(String[] args) throws IOException, TimeoutException {
    		// 获取连接
    		Connection con = ConnectionUtils.getConnection();
    		// 从连接中创建通道
    		final Channel channel = con.createChannel();
    		// 声明队列
    		channel.queueDeclare(QUEUE, true, false, false, null);
    		// 绑定交换机,指定路由键为error,info,warning
    		channel.queueBind(QUEUE, EXCHANGE, "error");
    		channel.queueBind(QUEUE, EXCHANGE, "info");
    		channel.queueBind(QUEUE, EXCHANGE, "warning");
    		// 保证一次只分发一条消息
    		channel.basicQos(1);
    		// 创建消费者
    		Consumer consumer = new DefaultConsumer(channel) {
    			// 获取消息
    			@Override
    			public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
    					throws IOException {
    				String msg = new String(body, "utf-8");
    				System.out.println("Recver2接收到消息——" + msg);
    				// 手动确认消息
    				channel.basicAck(envelope.getDeliveryTag(), false);
    			}
    
    		};
    		// 监听队列
    		channel.basicConsume(QUEUE, false, consumer);
    	}
    }

先运行Sender创建交换机，在运行消费者和交换机，发现消费者1和2都接收到消息了，把生产者Sender的路由键改为info，则只有消费者2能接收到消息。

路由模式的缺陷：

必须明确的指定路由键，不支持匹配路由键，如果一个队列有很多个路由键这样写起来太麻烦，所以需要一种可以匹配的路由键，可以使用Topics模式来实现。

### Topics主题模式 ###

Topics支持将选择键与绑定键进行匹配。发送给主题转发器的消息不能是任意设置的选择键，必须是用小数点隔开的一系列的标识符。这些标识符可以是随意，但是通常跟消息的某些特性相关联。一些合法的路由选择键比如“socket.usd.nyse”,"nyse.vmw","quick.orange.rabbit"，你愿意用多少单词都可以，只要不超过上限的255个字节。

绑定键也必须以相同的格式。主题转发器的逻辑类似于direct类型的转发器。消息通过一个特定的路由键发送到所有与绑定键匹配的队列中。需要注意的是，关于绑定键有两种特殊的情况：

*  \#：匹配0个多个标识符。
 *  \*：匹配1个标识符。

一般\#常用。比如goods.\#可以匹配goods.add、goods.img.add、goods.img.update。。。而goods.\*只能匹配goods.add、goods.update。

![70 7][]

对于上面的模型图可以这么理解：

我们准备发送关于动物的消息，消息会附加一个选择键包含3个标识符（两个点隔开）。第一个标识符描述动物的速度，第二个标识符描述动物的颜色，第三个标识符描述动物的物种：<speed>.<color>.<species>。

我们创建3个绑定键：Q1与\*.orange.\*绑定Q2与\*.\*.rabbit和lazy.\#绑定。

可以简单的认为：

*  Q1对所有的橙色动物感兴趣。
 *  Q2想要知道关于兔子的一切以及关于懒洋洋的动物的一切。

一个附带quick.orange.rabbit的选择键的消息将会被转发到两个队列。附带lazy.orange.elephant的消息也会被转发到两个队列。另一方面quick.orange.fox只会被转发到Q1，lazy.brown.fox将会被转发到Q2。lazy.pink.rabbit虽然与两个绑定键匹配，但是也只会被转发到Q2一次。quick.brown.fox不能与任何绑定键匹配，所以会被丢弃。如果我们违法我们的约定，发送一个或者四个标识符的选择键，类似：orange，quick.orange.male.rabbit，这些选择键不能与任何绑定键匹配，所以消息将会被丢弃。另一方面，lazy.orange.male.rabbit，虽然是四个标识符，也可以与lazy.\#匹配，从而转发至Q2。

**生产者Sender**

public class Sender {
    	private static final String EXCHANGE = "test_exchange_topic";
    
    	public static void main(String[] args) {
    		Connection con = null;
    		Channel channel = null;
    		try {
    			// 获取连接
    			con = ConnectionUtils.getConnection();
    			// 从连接中创建通道
    			channel = con.createChannel();
    			// 声明交换机,采用topic类型
    			channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.TOPIC, false, false, null);
    			// 消息
    			String msg = "你好 exchange topic!";
    			// 发送消息
    			channel.basicPublish(EXCHANGE, "quick.orange.rabbit", null, msg.getBytes());
    			System.out.println("send success");
    		} catch (IOException e) {
    			e.printStackTrace();
    		} catch (TimeoutException e) {
    			e.printStackTrace();
    		} finally {
    			// 关闭连接
    			ConnectionUtils.close(channel, con);
    		}
    	}
    }

**消费者Recver1**，路由键为\*.orange.\*。

public class Recver1 {
    	private static final String EXCHANGE = "test_exchange_topic";
    	private static final String QUEUE = "test_queue_topic_1";
    
    	public static void main(String[] args) throws IOException, TimeoutException {
    		// 获取连接
    		Connection con = ConnectionUtils.getConnection();
    		// 从连接中创建通道
    		final Channel channel = con.createChannel();
    		// 声明队列
    		channel.queueDeclare(QUEUE, true, false, false, null);
    		// 绑定交换机,指定路由键为error
    		channel.queueBind(QUEUE, EXCHANGE, "*.orange.*");
    		// 保证一次只分发一条消息
    		channel.basicQos(1);
    		// 创建消费者
    		Consumer consumer = new DefaultConsumer(channel) {
    			// 获取消息
    			@Override
    			public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
    					throws IOException {
    				String msg = new String(body, "utf-8");
    				System.out.println("Recver1接收到消息——" + msg);
    				// 手动确认消息
    				channel.basicAck(envelope.getDeliveryTag(), false);
    			}
    
    		};
    		// 监听队列
    		channel.basicConsume(QUEUE, false, consumer);
    	}
    }

**消费者Recver2**，路由键为\*.\*.rabbit和lazy.\#。

public class Recver2 {
    	private static final String EXCHANGE = "test_exchange_topic";
    	private static final String QUEUE = "test_queue_topic_2";
    
    	public static void main(String[] args) throws IOException, TimeoutException {
    		// 获取连接
    		Connection con = ConnectionUtils.getConnection();
    		// 从连接中创建通道
    		final Channel channel = con.createChannel();
    		// 声明队列
    		channel.queueDeclare(QUEUE, true, false, false, null);
    		// 绑定交换机,指定路由键为error
    		channel.queueBind(QUEUE, EXCHANGE, "*.*.rabbit");
    		channel.queueBind(QUEUE, EXCHANGE, "lazy.#");
    		// 保证一次只分发一条消息
    		channel.basicQos(1);
    		// 创建消费者
    		Consumer consumer = new DefaultConsumer(channel) {
    			// 获取消息
    			@Override
    			public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
    					throws IOException {
    				String msg = new String(body, "utf-8");
    				System.out.println("Recver2接收到消息——" + msg);
    				// 手动确认消息
    				channel.basicAck(envelope.getDeliveryTag(), false);
    			}
    
    		};
    		// 监听队列
    		channel.basicConsume(QUEUE, false, consumer);
    	}
    }

先运行Sender创建交换机，在运行消费者和交换机，发现消费者1和2都接收到消息了，把生产者Sender的路由键改为quick.orange.fox，则只有消费者1能接收到消息，和上面分析的结果是一样的。

[70]: /images/20220508/ae6e952e348848478ccc8018453e33d0.png
[70 1]: /images/20220508/cf3ad9cf6792418bae1b755e5d7c7189.png
[70 2]: /images/20220508/84d621365f5a488a8be6f22cb8a6acce.png
[70 3]: /images/20220508/b45645011d444f1e84b234ede3ce67a7.png
[70 4]: /images/20220508/ea2f73b10a584a7cbc9d835a502f0fc2.png
[70 5]: /images/20220508/29f97a17421d4484919a8b34677f66c6.png
[70 6]: /images/20220508/20990681b0644bab939625ad1e4f3797.png
[70 7]: /images/20220508/11d762505d9b49f593557582036afd34.png