MQ的訂閱模式

一：介紹

1.模式

　　

 

2.使用場景

　　一個生產者，多個消費者

　　每一個消費者都有自己的隊列

　　生產者沒有直接把消息發送給隊列，而是發送到了交換機

　　每一個隊列都要綁定到交換機

　　可以實現一個消息被多個消費者消費。

 

二：程序

1.生產者

package com.mq.PubSubFanout;

import com.mq.utils.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

public class FanoutSend {
    private static final String EXCHANGE_NAME="text_exchange_fanout";
    public static void main(String[] args) throws Exception {
        //獲取一個連接
        Connection connection= ConnectionUtil.getConnection();
        //從連接中獲取一個通道
        Channel channel=connection.createChannel();
        //創建交換機
        channel.exchangeDeclare(EXCHANGE_NAME,"fanout");

        //消息
        String msg="hello pubsub";

        //發送
        channel.basicPublish(EXCHANGE_NAME,"",null,msg.getBytes());

        System.out.println("send msg:"+msg);
        //關閉連接
        channel.close();
        connection.close();
    }
}

 

2.消費者一

package com.mq.PubSubFanout;

import com.mq.utils.ConnectionUtil;
import com.rabbitmq.client.*;

import java.io.IOException;

public class FanoutReceive1 {
    private static final String EXCHANGE_NAME="text_exchange_fanout";
    private static final String QUENE_NAME="test_fanout_queue_email";
    public static void main(String[] args)throws Exception{
        //獲取一個連接
        Connection connection = ConnectionUtil.getConnection();
        //創建通道
        final Channel channel = connection.createChannel();
        //創建隊列聲明
        channel.queueDeclare(QUENE_NAME,false,false,false,null);

        //綁定交換機
        channel.queueBind(QUENE_NAME,EXCHANGE_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]receive msg:"+msg);
                try {
                    Thread.sleep(200);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }finally {
                    System.out.println("done");
                    //手動應答
                    channel.basicAck(envelope.getDeliveryTag(),false);
                }
            }
        };
        //監聽隊列,不是自動應答
        boolean autoAck=false;
        channel.basicConsume(QUENE_NAME,autoAck,consumer);
    }
}

 

3.消費者二

package com.mq.PubSubFanout;

import com.mq.utils.ConnectionUtil;
import com.rabbitmq.client.*;

import java.io.IOException;

public class FanoutReceive2 {
    private static final String EXCHANGE_NAME="text_exchange_fanout";
    private static final String QUENE_NAME="test_fanout_queue_ems";
    public static void main(String[] args)throws Exception{
        //獲取一個連接
        Connection connection = ConnectionUtil.getConnection();
        //創建通道
        final Channel channel = connection.createChannel();
        //創建隊列聲明
        channel.queueDeclare(QUENE_NAME,false,false,false,null);

        //綁定交換機
        channel.queueBind(QUENE_NAME,EXCHANGE_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]receive msg:"+msg);
                try {
                    Thread.sleep(200);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }finally {
                    System.out.println("done");
                    //手動應答
                    channel.basicAck(envelope.getDeliveryTag(),false);
                }
            }
        };
        //監聽隊列,不是自動應答
        boolean autoAck=false;
        channel.basicConsume(QUENE_NAME,autoAck,consumer);
    }
}

 

4.效果

　　send：

　　

　　receive1：

　　

　　receive2：

　　

 

5.運行注意點

　　如果之間運行receive類，會發現報錯，因為沒有交換機。

　　所以，可以先運行send類，雖然交換機不能存儲發送的消息，但是可以創建交換機。

　　然后，就可以按照原來的方式。

　　先啟動兩個消費者進行監聽，然后啟動生產者。

　　現象：就是消費者都獲取到了生產者生產的消息。