简书：亚武de小文 【原创：转载请注明出处】
路由模式（Routing）
Direct类型交换机
LengToo上学.png
RabbitMQ有以下几种工作模式 ：
- Work queues
- Publish/Subscribe
- Routing
- Topic
- Headers
- RPC
Routing
模型图
[亚武de小文]Routing模型图.png
Prouducer：生产者，是消息的发出者。
Exchanger：direct类型的交换机，此处为direct，它负责接收生产者的消息，并根据路由键分发消息到指定的队列。
Queue1、Queue2：队列。Q1队列只会接收：消息路由键为health的消息，Q2队列只会接收：消息路由键为mentality和deliciousfood的消息。
Consumer1、Consumer2：消费者，它们负责从队列中获取消息并消费。
routing-key：health，mentality，deliciousfood
- 路由模式：
1、每个消费者监听自己的队列，并且设置routingkey。
2、生产者将消息发给交换机，由交换机根据routingkey来转发消息到指定的队列。
参考代码
生产者
-
Producer.javapackage com.yawu.xiaowen.direct; import com.rabbitmq.client.BuiltinExchangeType; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** * Direct交换机 * 生产者-发件人01 * * @author yawu * @date 2019.06.26 */ public class Producer { private static final Logger LOGGER = LoggerFactory.getLogger(Producer.class); private static final String EXCHANGE_NAME = "mq_direct_exchange"; private static final String QUEUE_NAME_01 = "mq_direct_queue_01"; private static final String QUEUE_NAME_02 = "mq_direct_queue_02"; /** * 路由键:健康 */ private static final String ROUTING_KEY_HEALTH = "routekey_health"; /** * 路由键:心态 */ private static final String ROUTING_KEY_MENTALITY = "routekey_mentality"; /** * 路由键:美食 */ private static final String ROUTING_KEY_DELICIOUSFOOD = "routekey_deliciousfood"; public static void main(String[] args) { try { // RabbitMQ建立连接的管理器。 ConnectionFactory factory = new ConnectionFactory(); // 设置服务器地址 factory.setHost("127.0.0.1"); factory.setUsername("guest"); factory.setPassword("guest"); // 创建一个连接 Connection connection = factory.newConnection(); // 创建一个信道 Channel channel = connection.createChannel(); // 声明一个Direct类型的交换机 channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT); // 声明两个个队列 /** * 【参数详解】 * queue：要创建的队列名 * durable：是否持久化。如果为true，可以在RabbitMQ崩溃后恢复消息 * exclusive：true表示一个队列只能被一个消费者占有并消费 * autoDelete：true表示服务器不在使用这个队列是会自动删除它 * arguments：其它参数 */ channel.queueDeclare(QUEUE_NAME_01, true, false, false, null); channel.queueDeclare(QUEUE_NAME_02, true, false, false, null); // 开始绑定订阅 /** * 【参数详解】 * 队列一对健康感兴趣 * 队列对心态和美食感兴趣 * queue：要被绑定的队列名 * exchange：队列绑定到哪个路由器上 * routingKey：队列对这种路由键感兴趣，路由器会把这种routingKey的消息发送给队列 */ // 队列一绑定或订阅了交换机关于健康的消息 channel.queueBind(QUEUE_NAME_01, EXCHANGE_NAME, ROUTING_KEY_HEALTH); // 队列二绑定或订阅了交换机关于心态和美食的消息 channel.queueBind(QUEUE_NAME_02, EXCHANGE_NAME, ROUTING_KEY_MENTALITY); channel.queueBind(QUEUE_NAME_02, EXCHANGE_NAME, ROUTING_KEY_DELICIOUSFOOD); // 模拟发送消息 String message01 = "明天有健康讲座"; String message02 = "人生就像一场戏，因为有缘才相距"; String message03 = "海底捞火锅"; // 向交换机发送3个消息 channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY_HEALTH, null, message01.getBytes("UTF-8")); channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY_MENTALITY, null, message02.getBytes("UTF-8")); channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY_DELICIOUSFOOD, null, message03.getBytes("UTF-8")); LOGGER.info("消息发送成功"); channel.close(); connection.close(); } catch (Exception e) { e.printStackTrace(); } } } -
Consumner01.javapackage com.yawu.xiaowen.direct; import com.rabbitmq.client.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.IOException; /** * Direct交换机 * 消费者-收件人01 * * @author yawu * @date 2019.06.26 */ public class Consumer01 { private static final Logger LOGGER = LoggerFactory.getLogger(Producer.class); private static final String EXCHANGE_NAME = "mq_direct_exchange"; private static final String QUEUE_NAME_01 = "mq_direct_queue_01"; private static final String QUEUE_NAME_02 = "mq_direct_queue_02"; /** * 路由键:健康 */ private static final String ROUTING_KEY_HEALTH = "routekey_health"; /** * 路由键:心态 */ private static final String ROUTING_KEY_MENTALITY = "routekey_mentality"; /** * 路由键:美食 */ private static final String ROUTING_KEY_DELICIOUSFOOD = "routekey_deliciousfood"; public static void main(String[] args) { try { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("127.0.0.1"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); // 声明一个Direct类型的交换机 channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT); // 声明两个队列 channel.queueDeclare(QUEUE_NAME_01, true, false, false, null); // 开始绑定 // 队列一绑定或订阅了交换机关于健康的消息 channel.queueBind(QUEUE_NAME_01, EXCHANGE_NAME, ROUTING_KEY_HEALTH); // 模拟接收消息 Consumer consumer = new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { final String message = new String(body, "UTF-8"); LOGGER.info("队列一收到消息:{}", message); } }; // 队列一确认收到消息 channel.basicConsume(QUEUE_NAME_01, true, consumer); } catch (Exception e) { e.printStackTrace(); } } } -
Consumer02.javapackage com.yawu.xiaowen.direct; import com.rabbitmq.client.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.IOException; /** * Direct交换机 * 消费者-收件人02 * * @author yawu * @date 2019.06.26 */ public class Consumer02 { private static final Logger LOGGER = LoggerFactory.getLogger(Producer.class); private static final String EXCHANGE_NAME = "mq_direct_exchange"; private static final String QUEUE_NAME_01 = "mq_direct_queue_01"; private static final String QUEUE_NAME_02 = "mq_direct_queue_02"; /** * 路由键:健康 */ private static final String ROUTING_KEY_HEALTH = "routekey_health"; /** * 路由键:心态 */ private static final String ROUTING_KEY_MENTALITY = "routekey_mentality"; /** * 路由键:美食 */ private static final String ROUTING_KEY_DELICIOUSFOOD = "routekey_deliciousfood"; public static void main(String[] args) { try { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("127.0.0.1"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); // 声明一个Direct类型的交换机 channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT); // 声明队列 channel.queueDeclare(QUEUE_NAME_02, true, false, false, null); // 开始绑定 // 队列二绑定或订阅了交换机关于心态和美食的消息 channel.queueBind(QUEUE_NAME_02, EXCHANGE_NAME, ROUTING_KEY_MENTALITY); channel.queueBind(QUEUE_NAME_02, EXCHANGE_NAME, ROUTING_KEY_DELICIOUSFOOD); //模拟接收消息 Consumer consumer = new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { final String message = new String(body, "UTF-8"); LOGGER.info("队列二收到消息:{}", message); } }; //队列二确认收到消息 channel.basicConsume(QUEUE_NAME_02, true, consumer); } catch (Exception e) { e.printStackTrace(); } } }
测试及运行分析
1、打开RabbitMQ的管理界面，观察交换机绑定情况：
Routing交换机绑定.png
-
队列一绑定的Health
队列一绑定.png -
队列二绑定的心态和美食
队列二绑定.png
2、启动消费者和生产者服务，可以看到消费者之接收到了自己订阅绑定的感兴趣的内容
启动服务.png
消费者启动.png
消费者一接收的队列一消息.png
消费者02接收的队列二消息.png
3、至此，Direct交换机学习完毕！
总结
Direct类型交换机：
就是根据路由键的Key进行发送。如果Key一致，路由器就把数据投递到这个队列中。Fanout类型交换机：
直接把队列绑定到路由器。路由器在收到消息后，直接把消息投递到队列中，不需要路由键。发送到fanout exchange的所有消息会被转发到与exchange绑定的所有queue，不需要处理路由
