路由模式

模式说明

  1. 队列与交换机的绑定,不能是任意绑定了,而是要指定一个 RoutingKey(路由 key)。
  2. 消息的发送方在向 Exchange 发送消息时,也必须指定消息的 RoutingKey。
  3. Exchange 不再把消息交给每一个绑定的队列,而是根据消息的 Routing Key 进行判断,只有队列的 Routingkey 与消息的 Routing key 完全一致,才会接收到消息。

10_路由模式.png

各组件的说明如下:

组件 描述
P 生产者,向 Exchange 发送消息,发送消息时,会指定一个 routing key。
X Exchange(交换机),接收生产者的消息,然后把消息递交给与 routing key 完全匹配的队列。
C1 消费者,其所在队列指定了需要 routing key 为 error 的消息。
C2 消费者,其所在队列指定了需要 routing key 为 info、error、warning 的消息。

代码实现

在编码上与 Publish/Subscribe 发布与订阅模式的区别是交换机的类型为:Direct,还有队列绑定交换机的时候需要指定 routing key。

生产者

package com.haicoder.producer; import com.rabbitmq.client.BuiltinExchangeType; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import java.io.IOException; import java.util.concurrent.TimeoutException; /** * 发送消息 */ public class Producer_Routing { public static void main(String[] args) throws IOException, TimeoutException { //1.创建连接工厂 ConnectionFactory factory = new ConnectionFactory(); //2. 设置参数 factory.setHost("106.15.72.229");//ip 默认值 localhost factory.setPort(5672); //端口 默认值 5672 factory.setVirtualHost("/haicoder");//虚拟机 默认值/ factory.setUsername("heima");//用户名 默认 guest factory.setPassword("heima");//密码 默认值 guest //3. 创建连接 Connection Connection connection = factory.newConnection(); //4. 创建Channel Channel channel = connection.createChannel(); /* exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments) 参数: 1. exchange:交换机名称 2. type:交换机类型 DIRECT("direct"),:定向 FANOUT("fanout"),:扇形(广播),发送消息到每一个与之绑定队列。 TOPIC("topic"),通配符的方式 HEADERS("headers");参数匹配 3. durable:是否持久化 4. autoDelete:自动删除 5. internal:内部使用。 一般false 6. arguments:参数 */ String exchangeName = "test_direct"; //5. 创建交换机 channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT,true,false,false,null); //6. 创建队列 String queue1Name = "test_direct_queue1"; String queue2Name = "test_direct_queue2"; channel.queueDeclare(queue1Name,true,false,false,null); channel.queueDeclare(queue2Name,true,false,false,null); //7. 绑定队列和交换机 /* queueBind(String queue, String exchange, String routingKey) 参数: 1. queue:队列名称 2. exchange:交换机名称 3. routingKey:路由键,绑定规则 如果交换机的类型为fanout ,routingKey设置为"" */ //队列1绑定 error channel.queueBind(queue1Name,exchangeName,"error"); //队列2绑定 info error warning channel.queueBind(queue2Name,exchangeName,"info"); channel.queueBind(queue2Name,exchangeName,"error"); channel.queueBind(queue2Name,exchangeName,"warning"); String body = "日志信息:张三调用了delete方法...出错误了。。。日志级别:error..."; //8. 发送消息 channel.basicPublish(exchangeName,"warning",null,body.getBytes()); //9. 释放资源 channel.close(); connection.close(); } }

运行后,网页终端如下:

11_路由模式.png

消费者1

package com.haicoder.consumer; import com.rabbitmq.client.*; import java.io.IOException; import java.util.concurrent.TimeoutException; public class Consumer_Routing1 { public static void main(String[] args) throws IOException, TimeoutException { //1.创建连接工厂 ConnectionFactory factory = new ConnectionFactory(); //2. 设置参数 factory.setHost("106.15.72.229");//ip 默认值 localhost factory.setPort(5672); //端口 默认值 5672 factory.setVirtualHost("/haicoder");//虚拟机 默认值/ factory.setUsername("heima");//用户名 默认 guest factory.setPassword("heima");//密码 默认值 guest //3. 创建连接 Connection Connection connection = factory.newConnection(); //4. 创建Channel Channel channel = connection.createChannel(); String queue1Name = "test_direct_queue1"; String queue2Name = "test_direct_queue2"; /* basicConsume(String queue, boolean autoAck, Consumer callback) 参数: 1. queue:队列名称 2. autoAck:是否自动确认 3. callback:回调对象 */ // 接收消息 Consumer consumer = new DefaultConsumer(channel){ /* 回调方法,当收到消息后,会自动执行该方法 1. consumerTag:标识 2. envelope:获取一些信息,交换机,路由key... 3. properties:配置信息 4. body:数据 */ @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { /* System.out.println("consumerTag:"+consumerTag); System.out.println("Exchange:"+envelope.getExchange()); System.out.println("RoutingKey:"+envelope.getRoutingKey()); System.out.println("properties:"+properties);*/ System.out.println("body:"+new String(body)); System.out.println("将日志信息打印到控制台....."); } }; channel.basicConsume(queue1Name,true,consumer); //关闭资源?不要 } }

消费者2

package com.haicoder.consumer; import com.rabbitmq.client.*; import java.io.IOException; import java.util.concurrent.TimeoutException; public class Consumer_Routing2 { public static void main(String[] args) throws IOException, TimeoutException { //1.创建连接工厂 ConnectionFactory factory = new ConnectionFactory(); //2. 设置参数 factory.setHost("106.15.72.229");//ip 默认值 localhost factory.setPort(5672); //端口 默认值 5672 factory.setVirtualHost("/haicoder");//虚拟机 默认值/ factory.setUsername("heima");//用户名 默认 guest factory.setPassword("heima");//密码 默认值 guest //3. 创建连接 Connection Connection connection = factory.newConnection(); //4. 创建Channel Channel channel = connection.createChannel(); String queue1Name = "test_direct_queue1"; String queue2Name = "test_direct_queue2"; /* basicConsume(String queue, boolean autoAck, Consumer callback) 参数: 1. queue:队列名称 2. autoAck:是否自动确认 3. callback:回调对象 */ // 接收消息 Consumer consumer = new DefaultConsumer(channel){ /* 回调方法,当收到消息后,会自动执行该方法 1. consumerTag:标识 2. envelope:获取一些信息,交换机,路由key... 3. properties:配置信息 4. body:数据 */ @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { /* System.out.println("consumerTag:"+consumerTag); System.out.println("Exchange:"+envelope.getExchange()); System.out.println("RoutingKey:"+envelope.getRoutingKey()); System.out.println("properties:"+properties);*/ System.out.println("body:"+new String(body)); System.out.println("将日志信息存储到数据库....."); } }; channel.basicConsume(queue2Name,true,consumer); //关闭资源?不要 } }

测试

启动所有消费者,然后使用生产者发送消息;在消费者对应的控制台可以查看到生产者发送对应 routing key 对应队列的消息;到达按照需要接收的效果。

在执行完测试代码后,其实到 RabbitMQ 的管理后台找到 Exchanges 选项卡,点击 direct_exchange 的交换机,可以查看到如下的绑定:

12_路由模式.png

总结

Routing 模式要求队列在绑定交换机时要指定 routing key,消息会转发到符合 routing key 的队列。