RabbitMQ之Exchange、Queue参数详解_rabbitmq queue type

1.先来介绍RabbitMQ中的成员

• Producer(生产者)： 将消息发送到Exchange
• Exchange(交换器)：将从生产者接收到的消息路由到Queue
• Queue(队列)：存放供消费者消费的消息
• BindingKey(绑定键)：建立Exchange与Queue之间的关系（个人看作是一种规则，也就是Exchange将什么样的消息路由到Queue）
• RoutingKey(路由键)：Producer发送消息与路由键给Exchange，Exchange将判断RoutingKey是否符合BindingKey，如何则将该消息路由到绑定的Queue
• Consumer(消费者)：从Queue中获取消息

下面是各个成员的作用图解

引入依赖

<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.6.0</version>
</dependency>
1
2
3
4
5

2.先来介绍Exchange

这里将着重于介绍Exchange和Queue的各个参数解释

先来看看Exchange中都有哪些属性

• exchange：名称
• type：类型
• durable：是否持久化，RabbitMQ关闭后，没有持久化的Exchange将被清除
• autoDelete：是否自动删除，如果没有与之绑定的Queue，直接删除
• internal：是否内置的，如果为true，只能通过Exchange到Exchange
• arguments：结构化参数

type会在后面一节说到，这里不会讲

Exchange.DeclareOk exchangeDeclare(String exchange,
           String type,
           boolean durable,
           boolean autoDelete,
           boolean internal,
           Map<String, Object> arguments) throws IOException;
1
2
3
4
5
6

下面这个类用于创建一个与RabbitMQ的Connection(连接)，该Connection用于创建Channel(信道)，Channel是消息读写的通道，也就是我们的操作都会在Channel的基础之上进行

package com.dfyang.rabbitmq;

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

public class RabbitmqConnectionFactory {
    private static ConnectionFactory factory = new ConnectionFactory();

    private static String HOST = "192.168.195.123";
    private static int PORT = 5672;
    private static String USERNAME = "root";
    private static String PASSWORD = "151310";

    static {
        factory.setHost(HOST);
        factory.setPort(PORT);
        factory.setUsername(USERNAME);
        factory.setPassword(PASSWORD);
    }

    public static Connection newConnection() {
        try {
            return factory.newConnection();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

2.1先使用最简单的参数构建Exchange
exchangeDeclare(String exchange, String type)

package com.dfyang.rabbitmq.eq0;

import com.dfyang.rabbitmq.RabbitmqConnectionFactory;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

public class Exchange {

    private static String EXCHANGE_NAME = "exchange.0";

    public static void main(String[] args) throws Exception {
    	//创建一个Connection连接
        Connection connection = RabbitmqConnectionFactory.newConnection();
        //开启Channel
        Channel channel = connection.createChannel();
        //创建一个Exchange，设置名称为exchange.0和类型为direct
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");
        //关闭资源
        channel.close();
        connection.close();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

进入RabbitMQ可视化界面可以看到，RabbitMQ已经为我们创建了exchange.0，类型为direct
不懂如何进入可视化界面的可以点这

2.2接下来是三个参数，也就是加上了是否持久化，同时保留先前两个参数的exchange.0，之前我们已经创建了exchange.0，那么我们再创建一次会怎样
exchangeDeclare(String exchange, String type, boolean durable)

package com.dfyang.rabbitmq.eq0;

import com.dfyang.rabbitmq.RabbitmqConnectionFactory;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

public class Exchange {

    private static String EXCHANGE_NAME0 = "exchange.0";
    private static String EXCHANGE_NAME1 = "exchange.1";
    private static String EXCHANGE_NAME2 = "exchange.2";

    public static void main(String[] args) throws Exception {
        Connection connection = RabbitmqConnectionFactory.newConnection();
        Channel channel = connection.createChannel();
        //创建一个Exchange，设置名称为exchange.1和类型为direct
        channel.exchangeDeclare(EXCHANGE_NAME0, "direct");
        //创建一个Exchange，设置名称为exchange.1和类型为direct，持久化为false
        channel.exchangeDeclare(EXCHANGE_NAME1, "direct", false);
        //创建一个Exchange，设置名称为exchange.2和类型为direct，持久化为true
        channel.exchangeDeclare(EXCHANGE_NAME2, "direct", true);
        channel.close();
        connection.close();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

运行成功，并没有报错，因为只要你设置的的设置是一样的，那么就不会报错，如果设置的不一样，那么就会报错，后面会进行验证

这里我们发现exchange.2多了一个D标识，这个D是durable也就是持久化，而exchange.0没有持久化，也就是默认非持久化

接下来验证这个持久化有什么作用
关闭rabbitmq
rabbitmqctl stop_app
启动rabbitmq
rabbitmqctl start_app
重新进入可视化界面，Exchange就只剩下持久化的了

2.3接下来是五个参数的
多了两个参数，autoDelete和arguments
exchangeDeclare(String exchange, String type, boolean durable, boolean autoDelete, Map<String, Object> arguments)

下面创建了两个Exchange
exchange.3自动删除为false
exchange.4自动删除为true
由于这里是没有绑定Queue的，那么exchange.4将在创建后就被删除掉？

package com.dfyang.rabbitmq.eq0;

import com.dfyang.rabbitmq.RabbitmqConnectionFactory;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

public class Exchange {

    private static String EXCHANGE_NAME0 = "exchange.3";
    private static String EXCHANGE_NAME1 = "exchange.4";

    public static void main(String[] args) throws Exception {
        Connection connection = RabbitmqConnectionFactory.newConnection();
        Channel channel = connection.createChannel();
        //创建一个Exchange，设置名称为exchange.3和自动删除为false
        channel.exchangeDeclare(EXCHANGE_NAME1, "direct", false, false, null);
        //创建一个Exchange，设置名称为exchange.4和自动删除为true
        channel.exchangeDeclare(EXCHANGE_NAME1, "direct", false, true, null);
        channel.close();
        connection.close();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

执行上面的代码

exchange.4还活的好好的，这是因为我们必须在绑定Queue之后再失去绑定才会被删除，否则为什么不直接抛异常，接下来进行验证
下面直接通过可视化工具创建一个名称为queue.4的Queue

点击添加之后，点击queue.3进行绑定，输入exchange.3以及RoutingKey（下一节会细讲），点击bind

我们可以看到已经建立了Binding

那么我们点击Unbind，发现exchange.4没了

下面来看看第五个参数arguments的作用，这里依旧是使用可视化工具创建，后一节会使用很多java代码
再我们使用可视化创建Exchange时下面的Arguments提供了一个Alternate exchange，这就是一个arguments（这里只会展示这一个）

alternate-exchange：如下图，如果我们设置Exchange0的alternate-exchange为Exchange1，那么当Producer发送RoutingKey为test1的消息，Exchange0由于路由不到匹配的队列，会将消息发送给Exchange1进行路由

上面那张图我们创建了test.exchange.0，接下来创建test.exchange.1

然后创建test.queue.0和test.queue.1两个队列

将test.exchange.0绑定test.queue.0

将test.exchange.1绑定test.queue.1

点击test.exchange.0来发布一条消息，之一RoutingKey为test.1，也就是test.exchange.0路由不到匹配的Queue，将交给test.exchange.1进行路由，路由到匹配的test.queue.1

结果我们发现test.queue.1多了一条消息

2.4接下来是六个参数的
internal：是否内置的，如果为true，只能通过Exchange到Exchange
也就是该Producer无法直接将消息发送给该Exchange，只能将消息发送给一个Exchange再路由到该Exchange

直接使用可视化界面创建Exchange0设置internal为true
我们发现根本就没有publish这一功能，也就是我们无法通过Producer发布消息到该Exchange上

创建一个Queue0

绑定其与Exchange0的关系

创建一个Exchange1

绑定Exchange1与Exchange0的关系

使用Exchange1发送一条RoutingKey为test的消息，消息将被发送到Exchange0，再到Queue0

结果如下，至此，Exchange的6个参数全部讲解

2.5讲解完Exchange的参数，再来看Queue的参数，就会发现只有一个exclusive未讲
queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,
Map<String, Object> arguments

exclusive：是否排他，如果未true，则只在第一次创建它的Connection中有效，当Connection关闭，该Queue也会被删除

在执行完下面代码，查看可视化界面，发现queue中并没有exclusive.queue，因为在connection关闭后，该queue也会自动删除

package com.dfyang.rabbitmq.eq1;

import com.dfyang.rabbitmq.RabbitmqConnectionFactory;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

public class Exchange {
    private static String QUEUE_NAME0 = "exclusive.queue";

    public static void main(String[] args) throws Exception {
        Connection connection = RabbitmqConnectionFactory.newConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME0, true, true, false, null);
        channel.close();
        connection.close();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

修改代码，让代码一直听着

package com.dfyang.rabbitmq.eq1;

import com.dfyang.rabbitmq.RabbitmqConnectionFactory;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

public class Exchange {
    private static String QUEUE_NAME0 = "exclusive.queue";

    public static void main(String[] args) throws Exception {
        Connection connection = RabbitmqConnectionFactory.newConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME0, true, true, false, null);
        Thread.sleep(100000);
        channel.close();
        connection.close();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

打开可视化界面，点击exclusive.queue，再点击删除
翻译为：无法获得对锁定队列“独占”的独占访问。队列’在vhost ‘/’。它可以最初在另一个连接上声明，或者独占属性值与原始声明值不匹配。

完！

Linux安装RabbitMQ
RabbitMQ之Exchange类型详解