rabbitmq03 订阅模式--广播模式

一、

而在订阅模型中，多了一个exchange角色，而且过程略有变化：
P：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给X（交换机）
C：消费者，消息的接受者，会一直等待消息到来。
Queue：消息队列，接收消息、缓存消息。
Exchange：交换机，图中的X。一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给
某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。Exchange有常
见以下3种类型：
Fanout：广播，将消息交给所有绑定到交换机的队列
Direct：定向，把消息交给符合指定routing key 的队列
Topic：通配符，把消息交给符合routing pattern（路由模式） 的队列
Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有
符合路由规则的队列，那么消息会丢失！

发布订阅模式与工作队列模式的区别

1、工作队列模式不用定义交换机，而发布/订阅模式需要定义交换机。
2、发布/订阅模式的生产方是面向交换机发送消息，工作队列模式的生产方是面向队列发送消息(底层使用默认交换
机)。
3、发布/订阅模式需要设置队列和交换机的绑定，工作队列模式不需要设置，实际上工作队列模式会将队列绑 定到
默认的交换机 。

二、案例需求

一个频道
一个交换机 ，（类型为Fanout：广播）
两个队列
一个生产者
两个消费者（一个消费者对应一个队列，不然就成了竞争关系）

三、代码

producter:

package com.xiaoxu.fanout;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

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

/**
 * 入门工程
 */
public class Producter {

    //定义两个消息队列
    static  final  String QUEUE_FANOUT_1 = "queue_fanout_1";
    static  final  String QUEUE_FANOUT_2 = "queue_fanout_2";
    //defining a exchange
    static final  String EXCHANGE_NAME = "exchange_fanout";

    public static void main(String[] args) throws IOException, TimeoutException {
        //1、创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("39.101.204.74");//设置主机地址
        connectionFactory.setPort(5672);//端口号
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");

        //2. 创建连接；（抽取一个获取连接的工具类）
        Connection connection = connectionFactory.newConnection();
        // 3. 创建频道；
        Channel channel = connection.createChannel();

        //statement exchange
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);

        //5. 声明队列；
        /**
         * 参数1：队列名称
         * 参数2：是否定义持久化队列
         * 参数3：是否独占本次连接
         * 参数4：是否在不使用的时候自动删除队列
         * 参数5：队列其它参数
         */
        channel.queueDeclare(QUEUE_FANOUT_1,true,false,false,null);
        channel.queueDeclare(QUEUE_FANOUT_2,true,false,false,null);
        // 5. 队列绑定交换机
        channel.queueBind(QUEUE_FANOUT_1, EXCHANGE_NAME, "");
        channel.queueBind(QUEUE_FANOUT_2, EXCHANGE_NAME, "");


        // 6. 监听队列

        /**
         * 参数1：交换机名称，如果没有指定则使用默认Default Exchage
         * 参数2：路由key,简单模式可以传递队列名称
         * 参数3：消息其它属性
         * 参数4：消息内容
         */

        for (int i=1;i<=10;i++){
            String msg = "我是一个消息。。。。。" + i;
            channel.basicPublish(EXCHANGE_NAME,"",null,msg.getBytes());
            channel.basicPublish(EXCHANGE_NAME,"",null,msg.getBytes());
            System.out.println(msg);
        }

        //7、关闭资源
        channel.close();
        connection.close();
    }


}

consumer1:

package com.xiaoxu.fanout;


import com.rabbitmq.client.*;
import com.xiaoxu.ConnectionUtis;
import com.xiaoxu.fanout.Producter;

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

public class Consumer1 {

    public static void main(String[] args) throws IOException, TimeoutException {

        Connection connection = ConnectionUtis.getConnection();
        //创建频道
        Channel channel = connection.createChannel();
        //statement exchange
        channel.exchangeDeclare(Producter.EXCHANGE_NAME,BuiltinExchangeType.FANOUT);
        // create queue
        channel.queueDeclare(com.xiaoxu.fanout.Producter.QUEUE_FANOUT_1,true,false,false,null);
        //bainding  queue and exchange
        channel.queueBind(Producter.QUEUE_FANOUT_1,Producter.EXCHANGE_NAME,"");


        //creat consumer and set information
        DefaultConsumer consumer1 = new DefaultConsumer(channel) {
            @Override
          public void handleDelivery(String consumerTag, Envelope envelope,
                                     AMQP.BasicProperties properties, byte[] body) throws IOException{

                //路由key
                System.out.println("路由key为：" + envelope.getRoutingKey());
                //交换机
                System.out.println("交换机为：" + envelope.getExchange());
                //消息id
                System.out.println("消息id为：" + envelope.getDeliveryTag());
                //收到的消息
                System.out.println("接收到的消息为：" + new String(body, "utf-8"));

                System.out.println("我是消费者一号。。。");
            }
        };

        //listen to the message
        channel.basicConsume(Producter.QUEUE_FANOUT_1,true,consumer1);


    }
}

consumer2:

package com.xiaoxu.fanout;


import com.rabbitmq.client.*;
import com.xiaoxu.ConnectionUtis;

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

public class Consumer2 {

    public static void main(String[] args) throws IOException, TimeoutException {

        Connection connection = ConnectionUtis.getConnection();
        //创建频道
        Channel channel = connection.createChannel();
        //statement exchange
        channel.exchangeDeclare(Producter.EXCHANGE_NAME,BuiltinExchangeType.FANOUT);
        // create queue
        channel.queueDeclare(Producter.QUEUE_FANOUT_2,true,false,false,null);
        //bainding  queue and exchange
        channel.queueBind(Producter.QUEUE_FANOUT_2,Producter.EXCHANGE_NAME,"");


        //creat consumer and set information
        DefaultConsumer consumer2 = new DefaultConsumer(channel) {
            @Override
          public void handleDelivery(String consumerTag, Envelope envelope,
                                     AMQP.BasicProperties properties, byte[] body) throws IOException{

                //路由key
                System.out.println("路由key为：" + envelope.getRoutingKey());
                //交换机
                System.out.println("交换机为：" + envelope.getExchange());
                //消息id
                System.out.println("消息id为：" + envelope.getDeliveryTag());
                //收到的消息
                System.out.println("接收到的消息为：" + new String(body, "utf-8"));

                System.out.println("我是消费者二号。。。");
            }
        };

        //listen to the message
        channel.basicConsume(Producter.QUEUE_FANOUT_2,true,consumer2);


    }
}

三、测试