RabbitMQ--交换机

谁借莪１个温暖的怀抱￠ 2024-03-29 15:26 68阅读 0赞

## 一、概念 ##

#### 1、概念 ####

RabbitMQ消息传递模型的核心思想是：生产者生产的消息从不会直接发送到队列。通常生产者甚至都不知道这些消息传递到了那些队列中。

生产者只能将消息发送到交换机，交换机一方面它接收来自生产者的消息，另一方面将他推入到队列中。交换机必须确切知道如何处理接收到消息，把消息放入特定队列、放入许多队列、丢弃。

#### 2、类型 ####

交换机类型：直接(direct)、主题(topic)、标题(headers)、扇出(fanout)。

#### 3、 无名交换机。 ####

此交换机是默认交换机，通过空字符串进行标识。在下列代码中的第一个参数就是交换机的名字。消息能路由发送到队列其实是由routingkey(bindingkey)绑定key指定的。

//发消息
    channel.basicPublish("",workerOne.QUEUE_NAME,null,msg.getBytes());

4、临时队列

每当我们连接到Rabbit时，我们都需要一个全新的空队列，为此我们可以创建一个具有随机名称的队列。其次一旦我们断开了消费者的链接，队列将自动删除。

创建临时队列方式：channel.queueDeclare().getQueue();

创建后在web界面显示:

![442268658c2b4a10a79060c52209d424.png][]

## 二、类型介绍 ##

#### 1、Fanout ####

此种类型就是将接受到的消息广播到它知道的所有队列中。

代码示例：

消费者

//消息接收者
    public class Accept {
        public static final String EXCHANGE_NAME="this_is_name";
    
        public static void main(String[] args) {
            //创建链接
            Connection connection = ConnectionUtil.getConnection();
            try {
                //创建信道
                Channel channel = connection.createChannel();
                //创建交换机
                channel.exchangeDeclare(EXCHANGE_NAME,"fanout");
                //创建临时队列
                String queue = channel.queueDeclare().getQueue();
                //交换机绑定队列。参数1：队列名。2：交换机名。3：routingkey
                channel.queueBind(queue,EXCHANGE_NAME,"");
    
                System.out.printf("消费者2准备接收消息\n");
    
                //消费者成功消费的回调
                DeliverCallback deliverCallback=(var1, var2)->{
                    System.out.println("接收到的消息："+new String(var2.getBody()));
                };
    
                //取消消费
                CancelCallback cancelCallback= var->{
                    System.out.println("取消消费消息");
                };
                /**
                 * 消费消息
                 * 1、队列名
                 * 2、消费后是否自动应答
                 * 3、消费者成功消费的回调
                 * 4、消费者取消消费回调
                 * */
                channel.basicConsume(queue,true,deliverCallback,cancelCallback);
    
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    
    }

生产者

//生产者
    public class FanoutProduct {
        public static void main(String[] args) {
            //创建链接
            Connection connection = ConnectionUtil.getConnection();
            try {
                //创建信道
                Channel channel = connection.createChannel();
                //控制台输入内容
                Scanner scanner=new Scanner(System.in);
                while (scanner.hasNext()){
                    //发消息
                    channel.basicPublish(Accept.EXCHANGE_NAME,"",null,scanner.next().getBytes(StandardCharsets.UTF_8));
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

测试结果：使用多线程启动多个消费者，在使用生产者发送消息，多个消费者都接收到生产者发送的消息 。

#### 2、Direct Exchange ####

在Fanout类型上进行升级，交换机发送消息时，只对绑定了相应routingkey的队列中去。

实例代码：

消费者\*2

//消费者1
    public class DirectConsumer01 {
        public static final String EXCHANGE_NAME="this_is_name";
    
        public static void main(String[] args) {
            //创建链接
            Connection connection = ConnectionUtil.getConnection();
            try {
                //创建信道
                Channel channel = connection.createChannel();
                //创建交换机
                channel.exchangeDeclare(EXCHANGE_NAME,"direct");
                //创建临时队列
                channel.queueDeclare("queue01",false,false,false,null);
                //交换机绑定队列。参数1：队列名。2：交换机名。3：routingkey
                //下列进行多重绑定
                channel.queueBind("queue01",EXCHANGE_NAME,"001");
                channel.queueBind("queue01",EXCHANGE_NAME,"002");
    
                System.out.printf("消费者2准备接收消息\n");
    
                //消费者成功消费的回调
                DeliverCallback deliverCallback=(var1, var2)->{
                    System.out.println("接收到的消息："+new String(var2.getBody()));
                };
    
                //取消消费
                CancelCallback cancelCallback= var->{
                    System.out.println("取消消费消息");
                };
                /**
                 * 消费消息
                 * 1、队列名
                 * 2、消费后是否自动应答
                 * 3、消费者成功消费的回调
                 * 4、消费者取消消费回调
                 * */
                channel.basicConsume("queue01",true,deliverCallback,cancelCallback);
    
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

public class DirectConsumer02 {
        public static final String EXCHANGE_NAME="this_is_name";
    
        public static void main(String[] args) {
            //创建链接
            Connection connection = ConnectionUtil.getConnection();
            try {
                //创建信道
                Channel channel = connection.createChannel();
                //创建交换机
                channel.exchangeDeclare(EXCHANGE_NAME,"direct");
                //创建临时队列
                channel.queueDeclare("queue02",false,false,false,null);
                //交换机绑定队列。参数1：队列名。2：交换机名。3：routingkey
                //下列进行多重绑定
                channel.queueBind("queue02",EXCHANGE_NAME,"003");
    
                System.out.printf("消费者2准备接收消息\n");
    
                //消费者成功消费的回调
                DeliverCallback deliverCallback=(var1, var2)->{
                    System.out.println("接收到的消息："+new String(var2.getBody()));
                };
    
                //取消消费
                CancelCallback cancelCallback= var->{
                    System.out.println("取消消费消息");
                };
                /**
                 * 消费消息
                 * 1、队列名
                 * 2、消费后是否自动应答
                 * 3、消费者成功消费的回调
                 * 4、消费者取消消费回调
                 * */
                channel.basicConsume("queue02",true,deliverCallback,cancelCallback);
    
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

生产者

public class DirectProduct {
        public static void main(String[] args) {
            //创建链接
            Connection connection = ConnectionUtil.getConnection();
            try {
                //创建信道
                Channel channel = connection.createChannel();
                //控制台输入内容
                Scanner scanner=new Scanner(System.in);
                while (scanner.hasNext()){
                    //发消息
                    channel.basicPublish(Accept.EXCHANGE_NAME,"001",null,scanner.next().getBytes(StandardCharsets.UTF_8));
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

测试结果：生产者生产的消息指挥被传输到对应routingkey的队列中，在上列代码中即routingkey为“001”的消费者接受到消息。

#### 3、Topics ####

此类型交换机的routingkey不能随意写。必须满足一定的要求，它必须是一个单词列表， 以点好分开，例子：”name.key.word“，单词列表总和不能超过255个字符。

在规则列表中：

”\*“：星号，可以代替一个单词。

"\#"：井号，可以替代零个或多个单词。

主要代码：

消费者

//创建交换机
     channel.exchangeDeclare(EXCHANGE_NAME,"topic");
     //交换机绑定队列。参数1：队列名。2：交换机名。3：routingkey
     //routingkey解释：第二个单词为key的接受消息
     channel.queueBind("queue01",EXCHANGE_NAME,"*.key.#");

生产者

//发消息 
    channel.basicPublish(Accept.EXCHANGE_NAME,"name.key.xxx",null,scanner.next().getBytes(StandardCharsets.UTF_8));

[442268658c2b4a10a79060c52209d424.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/28/55ef541ae76c407b97ff9ff249745ba1.png