RabbitMQ消息中间件(第二章)第四部分-笔记

　　Binging-绑定

•  Exchange和Exchange、Queue之间的连接关系
• Binging可以包含RoutingKey或者参数

  Queue-消息队列　　

• 消息队列，实际存储消息数据 
• Durability：是否持久化，Durable：是，Transient：否
• Auto delete：如选yes，代表最后一个监听被移除之后，该Queue会自动被删除

　　Message-消息

• 服务器和应用程序之间传送的数据

• 本质就是一段数据，由Properties和Payload（Body）组成
• 常用属性：delivery mode、headers（自定义属性）

　　Message-其他属性

• content-type、content-encoding、priority
• correlation_id、reply_to、expiration、message_id
• timestamp、type、user_id、app_id、cluster_id

　　Virtual host-虚拟主机

• 虚拟地址，用于进行逻辑隔离，最上层的消息路由
• 一个Virtual host里面可以有若干个Exchange和Queue
• 同一个Virtual host里面不能有相同名称的Exchange或Queue

　　以下demo

package com.cx.temp.common.rabbitmq.message;


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

import java.util.HashMap;
import java.util.Map;

/**
 * 生产端
 */
public class Procuder {

    public static void main(String[] args) throws Exception{
        try {
            //1 创建一个ConectionFacory
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("127.0.0.1");
            connectionFactory.setPort(5672);
            connectionFactory.setVirtualHost("/test001");
            connectionFactory.setUsername("root");
            connectionFactory.setPassword("123456");

            //2 通过连接工厂创建连接
            Connection connection = connectionFactory.newConnection();

            //3 通过connection创建一个Channel
            Channel channel = connection.createChannel();

            Map<String, Object> header = new HashMap<>();
            header.put("my1", "11111");
            header.put("my2", "22222");

            AMQP.BasicProperties properties = new AMQP.BasicProperties().builder()
                    .deliveryMode(2) //持久化，假设消息服务重启后，该消息还会存在
                    .contentEncoding("UTF-8")
                    .expiration("10000") //10秒过期
                    .headers(header) //附加自定义数据
                    .build();

            //4 通过channel发送数据
            //以下参数对应：exchange routingKey（队列名称）  props  body
            //假如exchange传空，rabbitMq默认机制走的是exchange里的【AMQP defalut】，此时routingKey = 消费者的队列名就会被消费
            for (int i = 0; i < 5; i++) {
                String msg = "Hello RabbitMQ!";
                channel.basicPublish("", "test01", properties, msg.getBytes());
            }

            //5 记得关闭相关的连接
            channel.close();
            connection.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}

package com.cx.temp.common.rabbitmq.message;

import com.cx.temp.common.rabbitmq.quickstart.QueueingConsumer;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.util.Map;

/**
 * 消费端
 */
public class Consumer {

    public static void main(String[] args) throws Exception {

        //1 创建一个ConectionFacory
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("127.0.0.1");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/test001");
        connectionFactory.setUsername("root");
        connectionFactory.setPassword("123456");

        //2 通过连接工厂创建连接
        Connection connection = connectionFactory.newConnection();

        //3 通过connection创建一个Channel
        Channel channel = connection.createChannel();

        //4 声明（创建）一个队列
        //以下参数：
        // queue队列名称
        // durable是否持久化,持久化后服务重启队列还在
        // exclusive 独占，该队列只有我能监听，相当于加了把锁，保证消息顺序消费
        // authDelete 如果脱离了exchange，队列会自动被删除
        // arguments 扩展参数
        String queueName = "test01";
        channel.queueDeclare(queueName, true, false, false, null);

        //5 创建消费者
        //QueueingConsumenr这个应该是5.x之前的经典写法。但是在4.x的版本QueueingConsumer被标记废止5.x被移除。移除的原因是什么呢？
        //原来QueueingConsumer内部用LinkedBlockingQueue来存放消息的内容，而LinkedBlockingQueue：一个由链表结构组成的有界队列，
        // 照先进先出的顺序进行排序 ，未指定长度的话，默认 此队列的长度为Integer.MAX_VALUE，那么问题来了，如果生产者的速度远远
        // 大于消费者的速度，也许没等到队列阻塞的条件产生（长度达到Integer.MAX_VALUE）内存就完蛋了，在老的版本你可以通过
        // 设置 rabbitmq的prefetch属性channel.basicQos(prefetch)来处理这个问题如果不设置可能出现内存问题
        // （比如因为网络问题只能向rabbitmq生产不能消费，消费者恢复网络之后就会有大量的数据涌入，出现内存问题,oom fgc等）。
        QueueingConsumer queueingConsumer = new QueueingConsumer(channel);

        //6 设置Channel
        // 队列名
        // autoAck 是否自动签收
        // callback 具体消费者对象
        channel.basicQos(1);
        channel.basicConsume(queueName, true, queueingConsumer);

        while(true) {
            // 7 获取消息
            QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();
            String msg = new String(delivery.getBody());
            System.out.println("消费端：" + msg);
            Map<String, Object> headers = delivery.getProperties().getHeaders();

            System.out.println("headrs get my1 value" + headers.get("my1"));

            //Envelope envelope = delivery.getEnvelope();
        }
    }

}