rabbitmq订阅模型-路由使用例子

青旅半醒 2021-09-24 14:22 347阅读 0赞

订阅-路由模型图  
![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2RoajE5OTE4MQ_size_16_color_FFFFFF_t_70][]  
特点：在路由模式（Direct）中，可以实现不同的消息被不同的队列消费，交换机不再将消息发送给所有绑定的队列，而是根据Routing Key将消息发送到指定的队列，队列在与交换机绑定时会设定一个Routing Key，而生产者发送的消息时也需要携带一个Routing Key，只有消费端携带的Routing Key与生产者相同时，才可以接受到消息。  
  
1.编写生产者代码

public class DirectProduct {
    	
    	public static final String EXCHANGE_NAME = "direct_exchange";
    
    	public static void main(String[] args) throws IOException, TimeoutException {
    		Connection conn = ConnectionUtils.getConn();
    		//声明信道
    		Channel channel = conn.createChannel();
    		// 声明exchange，指定类型为direct
    		channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT,true);
    		//指定routingKey，个人理解实际上就是过滤的作用，消费端与此匹配的则才可以接受到消息
    		String routingKey = "success";
    		for (int i = 0; i <10; i++) {
    			//消息内容
    			String message = "订阅路由模式"+i;
    			//发布消息
    			channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes());
    		}
    		//关闭
    		channel.close();
    		conn.close();
    	}
    }

2.编写消费端1，routingkey与生产端的一样

public class DirectConsome1 {
    	public static final String QUEUE_NAME = "direct_queue_1";
    	
    	public static void main(String[] args) throws IOException, TimeoutException {
    		Connection conn = ConnectionUtils.getConn();
    		Channel channel = conn.createChannel();
    		//消费者声明自己的队列
    		channel.queueDeclare(QUEUE_NAME, true, false, false, null);
    		// 声明exchange，指定类型为direct
    		channel.exchangeDeclare(DirectProduct.EXCHANGE_NAME, BuiltinExchangeType.DIRECT,true);
    		//与生产者的routingKey一样
    		String routingKey = "success";
    		//消费者将队列与交换机进行绑定
    		channel.queueBind(QUEUE_NAME, DirectProduct.EXCHANGE_NAME, routingKey);
    		//handleDelivery父类没有实现，需要自己写
    		DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
    			@Override
    			public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
    					throws IOException {
    				System.out.println("消费者1："+new String(body, "UTF-8"));
    			}
    		};
    		//声明消费者队列
    		channel.basicConsume(QUEUE_NAME, true,defaultConsumer);
    	}
    }

3.编写消费端2，routingkey与生产端的不一样，设置为fail

public class DirectConsome2 {
    	public static final String QUEUE_NAME = "direct_queue_2";
    	
    	public static void main(String[] args) throws IOException, TimeoutException {
    		Connection conn = ConnectionUtils.getConn();
    		Channel channel = conn.createChannel();
    		//消费者声明自己的队列
    		channel.queueDeclare(QUEUE_NAME, true, false, false, null);
    		// 声明exchange，指定类型为direct
    		channel.exchangeDeclare(DirectProduct.EXCHANGE_NAME, BuiltinExchangeType.DIRECT,true);
    		//routingKey=fail
    		String routingKey = "fail";
    		//消费者将队列与交换机进行绑定
    		channel.queueBind(QUEUE_NAME, DirectProduct.EXCHANGE_NAME, routingKey);
    		//handleDelivery父类没有实现，需要自己写
    		DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
    			@Override
    			public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
    					throws IOException {
    				System.out.println("消费者1："+new String(body, "UTF-8"));
    			}
    		};
    		//声明消费者队列
    		channel.basicConsume(QUEUE_NAME, true,defaultConsumer);
    	}
    }

4.生产端发送消息，查看消费端消费情况，只有routingkey与生产者匹配的消费者1收到了消息。

![2020081116321950.png][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2RoajE5OTE4MQ_size_16_color_FFFFFF_t_70]: /images/20210923/7e6d678c2d77400a881d1dc30166f86e.png
[2020081116321950.png]: /images/20210923/f5d4b0fb80fb409983364e80a8a9419b.png