rabbitMQ:绑定Exchange发送和接收消息（fanout）

编写fanout消息接收类

由于fanout是广播机制，所以需要先有接收方接收，发送方才不会丢失数据

1.编写接收类Receive1

package com.it.rabbitmq.fanout;
 
import com.rabbitmq.client.*;
 
import java.io.IOException;
import java.util.concurrent.TimeoutException;
 
public class Receive1 {
    public static void main(String[] args)  {
        //创建链接工厂对象
        ConnectionFactory factory=new ConnectionFactory();
        factory.setUsername("root");
        factory.setPassword("123456");
        factory.setHost("192.168.174.129");
        factory.setPort(5672);
 
        Connection connection=null;//定义链接对象
        Channel channel=null;//定义通道对象
 
        try {
            connection=factory.newConnection();//实例化链接对象
            channel=connection.createChannel();//实例化通道对象
            /**
             * 由于fanout类型的交换机的消息是类似于广播的模式，它不需要绑定RoutingKey
             * 而又有可能会有很多个消费来接收这个交换机中的数据
             * 因此我们创建队列时，要创建一个随机的队列名称
             * 没有参数的方法会创建一个名称随机的队列
             * 这个队列的数据是非持久化的，是排外的，自动删除的
             */
            String name=channel.queueDeclare().getQueue();
            channel.exchangeDeclare("fanoutExchange", "fanout", true);
            channel.queueBind(name,"fanoutExchange","");
 
 
            channel.basicConsume(name,true, "",new DefaultConsumer(channel) {
 
                public void handleDelivery(String consumerTag,
                                           Envelope envelope,
                                           AMQP.BasicProperties properties,
                                           byte[] body) throws IOException {
                    //获取消息数据
                    String bodyStr = new String(body);
                    System.out.println("消费者1："+bodyStr);
                }
            });
 
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
 
    }
}

编写接收类Receive2

package com.it.rabbitmq.fanout;
 
import com.rabbitmq.client.*;
 
import java.io.IOException;
import java.util.concurrent.TimeoutException;
 
public class Receive2 {
    public static void main(String[] args)  {
        //创建链接工厂对象
        ConnectionFactory factory=new ConnectionFactory();
        factory.setUsername("root");
        factory.setPassword("123456");
        factory.setHost("192.168.174.129");
        factory.setPort(5672);
 
        Connection connection=null;//定义链接对象
        Channel channel=null;//定义通道对象
 
        try {
            connection=factory.newConnection();//实例化链接对象
            channel=connection.createChannel();//实例化通道对象
            /**
             * 由于fanout类型的交换机的消息是类似于广播的模式，它不需要绑定RoutingKey
             * 而又有可能会有很多个消费来接收这个交换机中的数据
             * 因此我们创建队列时，要创建一个随机的队列名称
             * 没有参数的方法会创建一个名称随机的队列
             * 这个队列的数据是非持久化的，是排外的，自动删除的
             */
            String name=channel.queueDeclare().getQueue();
            channel.exchangeDeclare("fanoutExchange", "fanout", true);
            channel.queueBind(name,"fanoutExchange","");
 
 
            channel.basicConsume(name,true, "",new DefaultConsumer(channel) {
 
                public void handleDelivery(String consumerTag,
                                           Envelope envelope,
                                           AMQP.BasicProperties properties,
                                           byte[] body) throws IOException {
                    //获取消息数据
                    String bodyStr = new String(body);
                    System.out.println("消费者2："+bodyStr);
                }
            });
 
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
 
    }
}

2.分别启动这两个接收类

 

3.进入控制台查看交换机和队列

 

 

 

注意：

1、使用fanout模式获取消息时不需要绑定特定的队列名称，只需使用channel.queueDeclare().getQueue();获取一个随机的队列名称，然后绑定到指定的Exchange即可获取消息。

2、这种模式中可以同时启动多个接收者只要都绑定到同一个Exchang即可让所有接收者同时接收同一个消息是一种广播的消息机制

编写fanout消息发送类

1.发送类

package com.it.rabbitmq.exchange.fanout;
 
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 Send {
    public static void main(String[] args) {
        //创建连接工厂
        ConnectionFactory factory=new ConnectionFactory();
        //配置rabbitMQ的连接信息
        factory.setHost("192.168.174.129");
        factory.setPort(5672);
        factory.setUsername("root");
        factory.setPassword("123456");
        //定义连接
        Connection connection=null;
        //定义通道
        Channel channel=null;
 
        try {
            connection=factory.newConnection();
            channel=connection.createChannel();
            /**
             * 由于使用fanout类型的交换机，因此消息的接收方可能会有多个
             * 因此不建议在消息发送时来创建队列，以及绑定交换机
             * 但是发送消息时，至少保证交换机是存在的
             */
//            channel.queueDeclare("myDirectQueue",true,false,false,null);
            channel.exchangeDeclare("fanoutExchange","fanout",true);
//            channel.queueBind("myDirectQueue","directExchange","directRoutingKey");
            String message="fanout的测试消息！";
            channel.basicPublish("fanoutExchange","",null,message.getBytes("utf-8"));
            System.out.println("消息发送成功，direct");
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }finally {
            if (channel!=null){
                try {
                    channel.close();
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (TimeoutException e) {
                    e.printStackTrace();
                }
            }
            if (connection!=null){
                try {
                    connection.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
 
        }
 
 
    }
}

2.测试发送

3.查看两个接收类的情况

注意：

fanout模式的消息需要将一个消息同时绑定到多个队列中因此这里不能创建并指定某个队列