RabbitMQ-订阅模式publish/subscribe

小鱼儿 2022-05-30 10:52 200阅读 0赞

### 文章目录 ###

*  RabbitMQ-订阅模式publish/subscribe
 *   *   *   *  1.订阅模式模型
             *   *  生产者
                 *  消费者1
                 *  消费者2

# RabbitMQ-订阅模式publish/subscribe #

#### 1.订阅模式模型 ####

![这里写图片描述][70]

> 1.  一个生产者 多个消费者
> 2.  每个消费者都有自己的队列
> 3.  生产者没有直接把消息发送给队列，而是先发送给交换机exchange
> 4.  每个队列都要绑定到交换机上
> 5.  生产者发送的消息是经过交换机的，然后到达队列，就能实现一个消息被多个消费者消费

>  应用场景: 比如注册之后需要发送邮件 同时需要发送短信

##### 生产者 #####

package com.ithzk.rabbitmq.ps;
    
    import com.ithzk.rabbitmq.utils.RabbitMQConnectionUtils;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * @author hzk * @date 2018/3/10 */
    public class Send { 
    
        private final static String EXCHANGER_NAME = "test_exchange_fanout";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
    
            Connection connection = RabbitMQConnectionUtils.getConnection();
    
            Channel channel = connection.createChannel();
    
            // 分发
            channel.exchangeDeclare(EXCHANGER_NAME,"fanout");
    
            String msg = "hello exchange";
    
            channel.basicPublish(EXCHANGER_NAME,"",null,msg.getBytes());
    
            System.out.println("Send msg"+msg);
    
            channel.close();
            connection.close();
        }
    
    }

![这里写图片描述][70 1]

>  消息丢失了，因为交换机没有存储的能力，rabbitMQ中只有队列有存储能力，此时还没有队列绑定到该交换机上，所以数据丢失了。

##### 消费者1 #####

package com.ithzk.rabbitmq.ps;
    
    import com.ithzk.rabbitmq.utils.RabbitMQConnectionUtils;
    import com.rabbitmq.client.*;
    
    import javax.sound.midi.Soundbank;
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * @author hzk * @date 2018/3/10 */
    public class Recv1 { 
    
        private static final String QUEUE_NAME="test_queue_fanout_email";
        private final static String EXCHANGER_NAME = "test_exchange_fanout";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
            //获取连接
            Connection connection = RabbitMQConnectionUtils.getConnection();
    
            //从连接中获取频道
            final Channel channel = connection.createChannel();
    
            //声明队列
            channel.queueDeclare(QUEUE_NAME,false,false,false,null);
    
            //绑定队列到交换机 转发器
            channel.queueBind(QUEUE_NAME,EXCHANGER_NAME,"");
    
            //保证一次只发一个
            channel.basicQos(1);
    
            DefaultConsumer consumer = new DefaultConsumer(channel) { 
    
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    String msg = new String(body, "utf-8");
                    System.out.println("[1] Recv msg:" + msg);
    
                    try { 
                        Thread.sleep(10000);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    } finally { 
                        System.out.println("[1] done");
                        channel.basicAck(envelope.getDeliveryTag(), false);
                    }
                }
            };
    
           boolean autoAck = false;
    
           channel.basicConsume(QUEUE_NAME,autoAck,consumer);
    
           System.out.println("[Consumer 1 start]");
    
    
        }
    
    
    }

##### 消费者2 #####

package com.ithzk.rabbitmq.ps;
    
    import com.ithzk.rabbitmq.utils.RabbitMQConnectionUtils;
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * @author hzk * @date 2018/3/10 */
    public class Recv2 { 
    
        private static final String QUEUE_NAME="test_queue_fanout_sms";
        private final static String EXCHANGER_NAME = "test_exchange_fanout";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
            //获取连接
            Connection connection = RabbitMQConnectionUtils.getConnection();
    
            //从连接中获取频道
            final Channel channel = connection.createChannel();
    
            //声明队列
            channel.queueDeclare(QUEUE_NAME,false,false,false,null);
    
            //绑定队列到交换机 转发器
            channel.queueBind(QUEUE_NAME,EXCHANGER_NAME,"");
    
            //保证一次只发一个
            channel.basicQos(1);
    
            DefaultConsumer consumer = new DefaultConsumer(channel) { 
    
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    String msg = new String(body, "utf-8");
                    System.out.println("[2] Recv msg:" + msg);
    
                    try { 
                        Thread.sleep(10000);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    } finally { 
                        System.out.println("[2] done");
                        channel.basicAck(envelope.getDeliveryTag(), false);
                    }
                }
            };
    
           boolean autoAck = false;
    
           channel.basicConsume(QUEUE_NAME,autoAck,consumer);
    
           System.out.println("[Consumer 2 start]");
    
    
        }
    
    
    }

![这里写图片描述][70 2]

>  和交换机绑定的队列都会收到消息

[70]: /images/20220530/12dccdef91194efcb3aa93cd4de7bda2.png
[70 1]: /images/20220530/962262f08bc7453986327150dbab499d.png
[70 2]: /images/20220530/a6899fe8aba348debdfbb8776335db96.png