当前位置：首页 > news >正文

RabbitMQ学习系列（四）：发布-订阅模型详解

news 来源：原创 2024/6/29 23:19:01

（一）发布-订阅模型（Publish/Subscribe）

发布订阅模型的结构图如下所示：

和前两个的模型结构不同，在发布订阅模型中多了一个X（exchange），exchange是一个交换机，生产者不是直接将消息发送给队列，而是先发送给交换机。消费者可以通过队列去订阅这个交换机，每个消费者对应于自己的一个队列。

这个结构就好像我们订阅微信公众号一样，作者将文章发送到自己的公众号上，只有订阅过该公众号的人才能收到消息。因此这个模型被称为发布-订阅模型。

（二）发布-订阅模型实践

发布订阅模型中多了交换机的存在，而我们在rabbitmq的可视化界面中就见到过exchange

继续通过代码展示该模型：

2.1 工具类

工具类和之前都一样，不做介绍了

public class ConnectionUtil {
    public static Connection getConnection() throws IOException, TimeoutException {
        //定义一个连接工厂
        ConnectionFactory factory=new ConnectionFactory();
        //设置服务地址
        factory.setHost("127.0.0.1");
        //设置AMQP端口
        factory.setPort(5672);
        //设置VHOSTS
        factory.setVirtualHost("/vhosts_sdxb");
        //设置用户名
        factory.setUsername("user_sdxb");
        factory.setPassword("123456");
        return factory.newConnection();
    }
}

2.2 生产者

public class Sent {
    private static final String EXCHANGE="ps_exchange";
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        //绑定交换机，设置类型为fanout
        channel.exchangeDeclare(EXCHANGE,"fanout");
        String msg="hello world";
        channel.basicPublish(EXCHANGE,"",null,msg.getBytes());
        channel.close();
        connection.close();
    }
}

生产者在发布订阅模型中不再绑定队列，而是绑定交换机。exchange的种类有4中，分别是Direct 、Fanout 、Topic、Headers。接下来会做详细介绍。

2.3 消费者一

public class Receive1 {
    private static final String QUEUE_NAME="ps_queue1";
    private static final String EXCHANGE="ps_exchange";
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionUtil.getConnection();
        final Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //将队列和交换机绑定
        channel.queueBind(QUEUE_NAME,EXCHANGE,"");
        //保证一次只分发一次
        channel.basicQos(1);
        Consumer 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(msg);
                //设置手动回执
                channel.basicAck(envelope.getDeliveryTag(),false);
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        //关闭自动回复
        boolean autoAck=false;
        channel.basicConsume(QUEUE_NAME,autoAck,consumer);
    }
}

消费者的代码和之前一样，唯一的区别是增加了队列和交换机的绑定channel.queueBind();

2.4 消费者二

public class Receive2 {
    private static final String QUEUE_NAME="ps_queue2";
    private static final String EXCHANGE="ps_exchange";
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionUtil.getConnection();
        final Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //将队列和交换机绑定
        channel.queueBind(QUEUE_NAME,EXCHANGE,"");
        //保证一次只分发一次
        channel.basicQos(1);
        Consumer 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(msg);
                //设置手动回执
                channel.basicAck(envelope.getDeliveryTag(),false);
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        //关闭自动回复
        boolean autoAck=false;
        channel.basicConsume(QUEUE_NAME,autoAck,consumer);
    }
}