6、RabbitMQ学习笔记—小结（Exchange、Queue、RoutingKey、Message、Vhostpath））_routingkey的"#.message.#

Binding

Exchange和Exchange、Queue之间的连接关系

• Binding中可以包含Routingkey或者参数

Queue

• 消息队列。实际存储消息数据
• Durability:是否持久化
• auto delete：yes，代表党最后一个监听被移除之后，该queue会自动删除

Message

• 服务器和应用程序之间传送的数据
• 本质上就是一段数据，由Properties和body组成
• 常用属性：delivery mode、headers、自定义属性、content_type、content_encoding、priority、reply_to、message_id、expiration、type、user_id…
  消息属性添加发送
  生产者

public static final String EXCHANGE_NAME = "test_exchange_message";
    public static final String ROUTING_KEY = "routingkey.message";
    public static void main(String[] args) throws IOException, TimeoutException {
        // 获取连接
        Connection connection = ConnectionsUtils.getConnection();
        // 创建信道
        Channel channel = connection.createChannel();
        // 发送消息
        String message = "test message for exchange_message";
        Map headers = new HashMap();
        headers.put("my1","111");
        headers.put("my2","222");
        AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
                .deliveryMode(2) // 2代表持久化消息，1：非持久化
                .expiration("6000") //过期时间
                .contentEncoding("utf-8")// 编码格式
                .headers(headers)
                .build();
        channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, properties, message.getBytes());
        System.out.println("发送成功：" + message);
        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

消费者

public static final String EXCHANGE_NAME = "test_exchange_message";
    public static final String EXCHANGE_TYPE = "direct";
    public static final String ROUTING_KEY = "routingkey.message";
    public static final String QUEUE_NAME = "test_queue_message";
    public static void main(String[] args) throws IOException, TimeoutException {
        // 获取连接
        Connection connection = ConnectionsUtils.getConnection();
        // 创建信道
        Channel channel = connection.createChannel();
        // 声明交换机
        channel.exchangeDeclare(EXCHANGE_NAME, EXCHANGE_TYPE, true, false, false ,null);
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 绑定
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);
        // 创建消费者进行监听获取消息
        DefaultConsumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("接收消息message: " + new String(body));
                Map map = properties.getHeaders();
                System.out.println(map.get("my1"));
                System.out.println(map.get("my2"));
            }
        };
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
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

启动消费者监听，进行消息发送，会在消费者控制台接受到消息和消息附带参数值

Virtual Host

虚拟主机

• 虚拟地址，用于逻辑隔离，最上层的消息路由
• 一个virtualhost 里边可以有若干个Exchange 和Queue
• 同一个virtualhost里边不能有相同名字的exchange和queue

小结