rabbitmq订阅模型-广播使用例子

短命女 2021-09-24 14:22 354阅读 0赞

订阅模型图  
![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2RoajE5OTE4MQ_size_16_color_FFFFFF_t_70][]  
特点：1个生产者，多个消费者；每一个消费者都有自己的一个队列；生产者没有将消息直接发送到队列，而是发送到了交换机；每个队列都要绑定到交换机；生产者发送的消息，经过交换机到达队列，实现一个消息被多个消费者获取的目的，生产者只负责发消息到交换机，至于消息该怎么发，以及发送到哪个队列，生产者都不负责。一般由消费者创建队列，并且绑定到交换机  
交换机的类型有以下几种:  
Fanout：广播，交换机将消息发送到所有与之绑定的队列中去  
Direct：定向，交换机按照指定的Routing Key发送到匹配的队列中去  
Topics：通配符，与Direct大致相同，不同在于Routing Key可以根据通配符进行匹配  
  
广播模式下，消息发送的流程如下：

1.可以有多个消费者，每个消费者都有自己的队列  
2.每个队列都要与exchange绑定  
3.生产者发送消息到exchange  
4.exchange将消息把消息发送到所有绑定的队列中去  
5.消费者从各自的队列中获取消息  
  
创建exchange  
![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2RoajE5OTE4MQ_size_16_color_FFFFFF_t_70 1][]  
创建两个队列，fanout\_queue\_1和fanout\_queue\_2  
![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2RoajE5OTE4MQ_size_16_color_FFFFFF_t_70 2][]  
1.编写生产者代码，生产者需要与交换机绑定，不需要管队列

public class FanoutProduct {
    	
    	public static final String EXCHANGE_NAME = "fanout_exchange";
    
    	public static void main(String[] args) throws IOException, TimeoutException {
    		Connection conn = ConnectionUtils.getConn();
    		//声明信道
    		Channel channel = conn.createChannel();
    		// 声明exchange，指定类型为fanout
    		channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT,true);
    		for (int i = 0; i <10; i++) {
    			//消息内容
    			String message = "订阅广播模式"+i;
    			//发布消息
    			channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
    		}
    		//关闭
    		channel.close();
    		conn.close();
    	}
    }

2.编写消费者1，指定队列fanout\_queue\_1，然后与生产这的交换机绑定

public class FanoutConsome1 {
    	public static final String QUEUE_NAME = "fanout_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，指定类型为fanout
    		channel.exchangeDeclare(FanoutProduct.EXCHANGE_NAME, BuiltinExchangeType.FANOUT,true);
    		//消费者将队列与交换机进行绑定
    		channel.queueBind(QUEUE_NAME, FanoutProduct.EXCHANGE_NAME, "");
    		//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，指定队列fanout\_queue\_2，然后与生产这的交换机绑定

public class FanoutConsome2 {
    	public static final String QUEUE_NAME = "fanout_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，指定类型为fanout
    		channel.exchangeDeclare(FanoutProduct.EXCHANGE_NAME, BuiltinExchangeType.FANOUT,true);
    		//消费者将队列与交换机进行绑定
    		channel.queueBind(QUEUE_NAME, FanoutProduct.EXCHANGE_NAME, "");
    		//handleDelivery父类没有实现，需要自己写
    		DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
    			@Override
    			public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
    					throws IOException {
    				System.out.println("消费者2："+new String(body, "UTF-8"));
    			}
    		};
    		//声明消费者队列
    		channel.basicConsume(QUEUE_NAME, true,defaultConsumer);
    	}
    }

4.启动两个消费者，然后再启动生产者，查看消费者接收消息情况  
![20200811160831718.png][]![20200811160847944.png][]  
  
两个消费者都接收到了相同消息

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2RoajE5OTE4MQ_size_16_color_FFFFFF_t_70]: /images/20210923/84857afa2ab945d69370c8abf01344b3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2RoajE5OTE4MQ_size_16_color_FFFFFF_t_70 1]: /images/20210923/8f78008e5896461d98bb9e4e86861ad7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2RoajE5OTE4MQ_size_16_color_FFFFFF_t_70 2]: /images/20210923/3d6b4e012aed4937be8f06f560fc34be.png
[20200811160831718.png]: https://img-blog.csdnimg.cn/20200811160831718.png
[20200811160847944.png]: /images/20210923/2c4e663f6ed44518913592101da691ce.png