RabbitMQ 之 订阅模式 Publish/Subscribe

模型图

我们之前学习的都是一个消息只能被一个消费者消费,那么如果我想发一个消息 能被多个消费者消费,这时候怎么办? 这时候我们就得用到了消息中的发布订阅模型

在前面的教程中，我们创建了一个工作队列，都是一个任务只交给一个消费者。这次我们做 将消息发送给多个消费者。这种模式叫做“发布/订阅”。

举列:    

类似微信订阅号 发布文章消息 就可以广播给所有的接收者。(订阅者)

那么咱们来看一下图,我们学过前两种有一些不一样,work 模式 是不是同一个队列 多个消费者,而 ps 这种模式呢,是一个队列对应一个消费者,pb 模式还多了一个 X(交换机 转发器) ,这时候我们要获取消息 就需要队列绑定到交换机上,交换机把消息发送到队列 , 消费者才能获取队列的消息

解读：
1、1 个生产者，多个消费者
2、每一个消费者都有自己的一个队列
3、生产者没有将消息直接发送到队列，而是发送到了交换机(转发器)
4、每个队列都要绑定到交换机
5、生产者发送的消息，经过交换机，到达队列，实现，一个消息被多个消费者获取的目的

生产者

package cn.wh.simple;

import java.io.IOException;
import java.util.concurrent.TimeoutException;


import cn.wh.util.RabbitMqConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

public class Send {

    private static final String  EXCHANGE_NAME="test_exchange_fanout";
    public static void main(String[] args) throws IOException, TimeoutException {
        
        Connection connection = RabbitMqConnectionUtil.getConnection();
        
        Channel channel = connection.createChannel();
        
        //声明交换机
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");//分发
        
        //发送消息
        String msg="hello ps";
        
        channel.basicPublish(EXCHANGE_NAME, "", null, msg.getBytes());
        
        System.out.println("Send :"+msg);
        
        channel.close();
        connection.close();
    }
}

消费者1

package cn.wh.simple;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import cn.wh.util.RabbitMqConnectionUtil;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.AMQP.BasicProperties;

public class Recv1 {
    
    private static final String QUEUE_NAME="test_queue_fanout_email";
    private static final String  EXCHANGE_NAME="test_exchange_fanout";
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = RabbitMqConnectionUtil.getConnection();
        final Channel channel = connection.createChannel();
        
        //队列声明
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        
        //绑定队列到交换机 转发器
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
        
        
        channel.basicQos(1);//保证一次只分发一个  
        
        //定义一个消费者
        Consumer consumer=new DefaultConsumer(channel){
            //消息到达 触发这个方法
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                    BasicProperties properties, byte[] body) throws IOException {
             
                String msg=new String(body,"utf-8");
                System.out.println("[1] Recv msg:"+msg);
                
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }finally{
                    System.out.println("[1] done ");
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        };
        
        boolean autoAck=false;//自动应答 false
        channel.basicConsume(QUEUE_NAME,autoAck , consumer);
    }
}

消费者2

package cn.wh.simple;

import java.io.IOException;
import java.util.concurrent.TimeoutException;


import cn.wh.util.RabbitMqConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.AMQP.BasicProperties;

public class Recv2 {
    
    private static final String QUEUE_NAME="test_queue_fanout_sms";
    private static final String  EXCHANGE_NAME="test_exchange_fanout";
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = RabbitMqConnectionUtil.getConnection();
        final Channel channel = connection.createChannel();

        //队列声明
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        //绑定队列到交换机 转发器
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
        channel.basicQos(1);//保证一次只分发一个

        //定义一个消费者
        Consumer consumer=new DefaultConsumer(channel){
            //消息到达 触发这个方法
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                    BasicProperties properties, byte[] body) throws IOException {

                String msg=new String(body,"utf-8");
                System.out.println("[2] Recv msg:"+msg);

                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }finally{
                    System.out.println("[2] done ");
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        };

        boolean autoAck=false;//自动应答 false
        channel.basicConsume(QUEUE_NAME,autoAck , consumer);
    }
}

 测试

一个消息 可以被多个消费者