RabbitMQ——訂閱模式類型/發布與訂閱模式

 

---

一、訂閱模式

訂閱模式示例圖：

 

 

 訂閱模型中，多了一個exchange角色：

• P：生產者，也就是要發送消息的程序，但是不再發送到隊列中，而是發給X（交換機）C：消費者，消息的接受者，會一直等待消息到來。

• Queue：消息隊列，接收消息、緩存消息。

• Exchange：交換機，圖中的X。一方面，接收生產者發送的消息。另一方面，知道如何處理消息，例如遞交給某個特別隊列、遞交給所有隊列、或是將消息丟棄。到底如何操作，取決於Exchange的類型。Exchange有常見以下3種類型：
  

  Fanout：廣播，將消息交給所有綁定到交換機的隊列
  Direct：定向，把消息交給符合指定routing key 的隊列
  Topic：通配符，把消息交給符合routing pattern（路由模式） 的隊列
  

Exchange（交換機）只負責轉發消息，不具備存儲消息的能力，因此如果沒有任何隊列與Exchange綁定，或者沒有符合路由規則的隊列，那么消息會丟失！ 

二、Publish/Subscribe發布與訂閱模式

1.模式說明

 發布訂閱模式：
（1）每個消費者監聽自己的隊列。
（2）生產者將消息發給broker，由交換機將消息轉發到綁定此交換機的每個隊列，每個綁定交換機的隊列都將接收到消息 

2、示例

Producer_PubSub:

/**
 * 發送消息
 */
public class Producer_PubSub {
    public static void main(String[] args) throws IOException, TimeoutException {

        //1.創建連接工廠
        ConnectionFactory factory = new ConnectionFactory();
        //2. 設置參數
        factory.setHost("172.16.98.133");//ip  默認值 localhost
        factory.setPort(5672); //端口  默認值 5672
        factory.setVirtualHost("/itcast");//虛擬機 默認值/
        factory.setUsername("jingdong");//用戶名 默認 guest
        factory.setPassword("jingdong");//密碼 默認值 guest
        //3. 創建連接 Connection
        Connection connection = factory.newConnection();
        //4. 創建Channel
        Channel channel = connection.createChannel();
       /*

       exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
       參數：
        1. exchange:交換機名稱
        2. type:交換機類型
            DIRECT("direct"),：定向,把消息交給符合指定routing key 的隊列。
            FANOUT("fanout"),：扇形（廣播），發送消息到每一個與之綁定隊列。
            TOPIC("topic"),通配符的方式
            HEADERS("headers");參數匹配

        3. durable:是否持久化
        4. autoDelete:自動刪除
        5. internal：內部使用。 一般false
        6. arguments：參數
        */

       String exchangeName = "test_fanout";
        //5. 創建交換機
        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.FANOUT,true,false,false,null);
        //6. 創建隊列
        String queue1Name = "test_fanout_queue1";
        String queue2Name = "test_fanout_queue2";
        channel.queueDeclare(queue1Name,true,false,false,null);
        channel.queueDeclare(queue2Name,true,false,false,null);
        //7. 綁定隊列和交換機
        /*
        queueBind(String queue, String exchange, String routingKey)
        參數：
            1. queue：隊列名稱
            2. exchange：交換機名稱
            3. routingKey：路由鍵，綁定規則
                如果交換機的類型為fanout ，routingKey設置為""
         */
        channel.queueBind(queue1Name,exchangeName,"");
        channel.queueBind(queue2Name,exchangeName,"");

        String body = "日志信息：張三調用了findAll方法...日志級別：info...";
        //8. 發送消息
        channel.basicPublish(exchangeName,"",null,body.getBytes());

        //9. 釋放資源
        channel.close();
        connection.close();

    }
}

Consumer_PubSub1:

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

        //1.創建連接工廠
        ConnectionFactory factory = new ConnectionFactory();
        //2. 設置參數
        factory.setHost("172.16.98.133");//ip  默認值 localhost
        factory.setPort(5672); //端口  默認值 5672
        factory.setVirtualHost("/itcast");//虛擬機 默認值/
        factory.setUsername("jingdong");//用戶名 默認 guest
        factory.setPassword("jingdong");//密碼 默認值 guest
        //3. 創建連接 Connection
        Connection connection = factory.newConnection();
        //4. 創建Channel
        Channel channel = connection.createChannel();


        String queue1Name = "test_fanout_queue1";
        String queue2Name = "test_fanout_queue2";


        /*
        basicConsume(String queue, boolean autoAck, Consumer callback)
        參數：
            1. queue：隊列名稱
            2. autoAck：是否自動確認
            3. callback：回調對象

         */
        // 接收消息
        Consumer consumer = new DefaultConsumer(channel){
            /*
                回調方法，當收到消息后，會自動執行該方法

                1. consumerTag：標識
                2. envelope：獲取一些信息，交換機，路由key...
                3. properties:配置信息
                4. body：數據

             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
              /*  System.out.println("consumerTag："+consumerTag);
                System.out.println("Exchange："+envelope.getExchange());
                System.out.println("RoutingKey："+envelope.getRoutingKey());
                System.out.println("properties："+properties);*/
                System.out.println("body："+new String(body));
                System.out.println("將日志信息打印到控制台.....");
            }
        };
        channel.basicConsume(queue1Name,true,consumer);


        //關閉資源？不要

    }
}

Consumer_PubSub2:

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

        //1.創建連接工廠
        ConnectionFactory factory = new ConnectionFactory();
        //2. 設置參數
        factory.setHost("172.16.98.133");//ip  默認值 localhost
        factory.setPort(5672); //端口  默認值 5672
        factory.setVirtualHost("/itcast");//虛擬機 默認值/
        factory.setUsername("jingdong");//用戶名 默認 guest
        factory.setPassword("jingdong");//密碼 默認值 guest
        //3. 創建連接 Connection
        Connection connection = factory.newConnection();
        //4. 創建Channel
        Channel channel = connection.createChannel();


        String queue1Name = "test_fanout_queue1";
        String queue2Name = "test_fanout_queue2";


        /*
        basicConsume(String queue, boolean autoAck, Consumer callback)
        參數：
            1. queue：隊列名稱
            2. autoAck：是否自動確認
            3. callback：回調對象

         */
        // 接收消息
        Consumer consumer = new DefaultConsumer(channel){
            /*
                回調方法，當收到消息后，會自動執行該方法

                1. consumerTag：標識
                2. envelope：獲取一些信息，交換機，路由key...
                3. properties:配置信息
                4. body：數據

             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
              /*  System.out.println("consumerTag："+consumerTag);
                System.out.println("Exchange："+envelope.getExchange());
                System.out.println("RoutingKey："+envelope.getRoutingKey());
                System.out.println("properties："+properties);*/
                System.out.println("body："+new String(body));
                System.out.println("將日志信息保存數據庫.....");
            }
        };
        channel.basicConsume(queue2Name,true,consumer);


        //關閉資源？不要

    }
}