rabbitmq订阅模型-主题使用例子

港控/mmm° 2021-09-24 14:22 328阅读 0赞

订阅模型-主题图  
![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2RoajE5OTE4MQ_size_16_color_FFFFFF_t_70][]  
特点：与路由模式基本类似，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符，Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割  
通配符规则：  
\#:匹配一个或多个词  
\*:匹配1个词  
  
1.编写生产者代码，，指定routingkey=success.update.customer

public class TopicProduct {
    	
    	public static final String EXCHANGE_NAME = "topic_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.TOPIC,true);
    		//指定routingKey，个人理解实际上就是过滤的作用，消费端与此匹配的则才可以接受到消息
    		String routingKey = "success.update.customer";
    		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=success.\#

public class TopicConsome1 {
    	public static final String QUEUE_NAME = "topic_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(TopicProduct.EXCHANGE_NAME, BuiltinExchangeType.TOPIC,true);
    		//与生产者的routingKey一样
    		String routingKey = "success.#";
    		//消费者将队列与交换机进行绑定
    		channel.queueBind(QUEUE_NAME, TopicProduct.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.编写消费者1代码，指定routingkey=success.\*

public class TopicConsome2 {
    	public static final String QUEUE_NAME = "topic_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(TopicProduct.EXCHANGE_NAME, BuiltinExchangeType.TOPIC,true);
    		//routingKey=fail
    		String routingKey = "success.*";
    		//消费者将队列与交换机进行绑定
    		channel.queueBind(QUEUE_NAME, TopicProduct.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.生产者发送消息，消费端查看消费情况，只有消费者1消费了消息，生产者的routing key是success.update.customer，二消费者2的routing key是success.\*，只能匹配到success.update，所有无法消费到消息  
![20200811170018214.png][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2RoajE5OTE4MQ_size_16_color_FFFFFF_t_70]: /images/20210923/c903c514360a4f368b0e2469a7e850fa.png
[20200811170018214.png]: /images/20210923/da5ce9abee6145d88d648ac429762fa2.png