RabbitMQ入门

标签：交换器入门 Exchange 队列 RabbitMQ Queue 消息

一、RabbitMQ简介安装

1、什么是MQ

MQ全称为Message Queue，即消息队列。“消息队列”是在消息的传输过程中保存消息的容器。它是典型的：生产者、消费者模型。生产者不断向消息队列中生产消息，消费者不断的从队列中获取消息。因为消息的生产和消费都是异步的，而且只关心消息的发送和接收，没有业务逻辑的侵入，这样就实现了生产者和消费者的解耦。

2、RabbitMQ

RabbitMQ 是一个在AMQP协议标准基础上完整的，可服用的企业消息系统。它遵循Mozilla Public License开源协议，采用 Erlang 实现的工业级的消息队列(MQ)服务器。

3、安装

3.1 上传安装包

3.2 安装Rabbit及依赖

cd /opt
#安装erlang
rpm -ivh erlang-21.3-1.el7.x86_64.rpm
#安装socat
yum install socat -y
#安装rabbitmq
rpm -ivh rabbitmq-server-3.8.8-1.el7.noarch.rpm

#添加开机启动rabbitmq服务
chkconfig rabbitmq-server on

开启web管理插件
rabbitmq-plugins enable rabbitmq_management

4、常用命令

#启动监控管理器
rabbitmq-plugins enable rabbitmq_management
#关闭监控管理器
rabbitmq-plugins disable rabbitmq_management

#启动rabbitmq：
systemctl start rabbitmq-service
#关闭rabbitmq：
systemctl stop  rabbitmq-service
#查看rabbitmq状态
systemctl status rabbitmq-service

#查看所有的队列：
rabbitmqctl list_queues
#关闭应用：
rabbitmqctl stop_app
#启动应用：
rabbitmqctl start_app

#添加用户
rabbitmqctl add_user admin 123
#设置用户角色
rabbitmqctl set_user_tags admin administrator
#设置用户权限
#用户 admin 具有/vhost1 这个 virtual host 中所有资源的配置、写、读权限
rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*"
#查看用户列表
rabbitmqctl list_users

二、基本使用

1、创建工具类

public class ConnectionUtil {

    /**
     * 获取连接
     * @param host rabbitMQ服务器地址
     * @param port rabbitMQ端口号
     * @param username 用户名
     * @param password 密码
     * @return
     * @throws Exception
     */
    public static Connection getConnection(String host,int port, String username,String password) throws Exception{
        //1.定义连接工厂
        ConnectionFactory factory = new ConnectionFactory();

        //2.设置服务器地址
        factory.setHost(host);

        //3.设置端口
        factory.setPort(port);

        //4.设置用户名、密码
        factory.setUsername(username);
        factory.setPassword(password);

        //5.通过工厂获取连接
        Connection connection = factory.newConnection();
        return connection;
    }
}

2、创建生产者，发布消息

public class ProducerTest {
    //队列名称
    private final static String QUEUE_NAME="hello";

    public static void main(String[] args) throws Exception {
        //1.获取连接
        Connection connection = ConnectionUtil.getConnection("192.168.10.100", ConnectionFactory.DEFAULT_AMQP_PORT,
                "admin", "123456");

        //2.声明通道
        Channel channel = connection.createChannel();

        //3.创建队列
        /**
         * queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,Map<String, Object> arguments)
             * queue: 队列名称
             * durable: 是否持久化，队列默认存放到内存中
             * exclusive: 是否独占连接，队列只允许在该连接中访问，如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时队列的创建
             * autoDelete: 是否自动删除
             * arguments: 设置其他参数
         */
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);

        //4.定义消息内容
        String message = "hello rabbitmq";

        //5.发布消息到exchange,同时指定路由规则
        /**
         * basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)
         * exchange: 指定交换机, ""表示使用默认交换机
         * routingKey: 指定路由规则，使用具体队列名称
         * props: 传递消息所携带的properties
         * body: 发布的具体消息,byte[]类型
         */
        channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
        System.out.println("发布消息完成!");
        //6.释放资源
        channel.close();
        connection.close();
    }
}

监控页面查看：

3、创建消费者，接收消息

public class ConsumerTest {
    private final static String QUEUE_NAME="hello";
    public static void main(String[] args) throws Exception {
        //1.获取连接
        Connection connection = ConnectionUtil.getConnection("192.168.10.100", ConnectionFactory.DEFAULT_AMQP_PORT,
                "admin", "123456");
        //2.声明通道
        Channel channel = connection.createChannel();
        //3.声明队列
        /**
         * queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,Map<String, Object> arguments)
             * queue: 队列名称
             * durable: 是否持久化，队列默认存放到内存中（设置false后MQ重启后队列全部删除）
             * exclusive: 是否独占连接，队列只允许在该连接中访问，如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时队列的创建
             * autoDelete: 是否自动删除
             * arguments: 设置其他参数
         */
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);

        //4.实现消费方法,这个方法类似事件监听，如果有消息的时候，会被自动调用
        DefaultConsumer consumer = new DefaultConsumer(channel){
	    /**
             * 当接收到消息后此方法被调用
             * @param consumerTag  标识消费者
             * @param envelope   信封
             * @param properties  消息属性
             * @param body  消息内容
             * @throws IOException
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("接收消息："+new String(body));
            }
        };

        /**
         *	消费者消费消息
         *	queue 队列名称
         *	autoAck 自动回复，当消费者接收到消息后要告诉mq消息已接收，如果将此参数设置为tru表示会自动回复mq
         *	callback，消费方法，当消费者接收到消息要执行的方法
         */
        channel.basicConsume(QUEUE_NAME,true,consumer);

        System.out.println("消费者开始监听队列!");
        System.in.read();

        //5.释放资源
        channel.close();
        connection.close();
    }
}

控制台打印：

三、RabbitMQ工作原理

1、基本架构图

Publisher - 生产者：发布消息到RabbitMQ中的Exchange
Consumer - 消费者：监听RabbitMQ中的Queue中的消息
Exchange - 交换机：和生产者建立连接并接收生产者的消息
Queue - 队列：Exchange会将消息分发到指定的Queue，Queue和消费者进行交互
Routes - 路由：交换机以什么样的策略将消息发布到Queue

2、工作原理图

各名词详解：

2.1 Producer（生产者）和 Consumer（消费者）

生产和消费消息。
消息一般由 2 部分组成：消息头 和 消息体。消息体也可以称为 payLoad ,消息体是不透明的，而消息头则由一系列的可选属性组成，这些属性包括 routing-key（路由键）、priority（相对于其他消息的优先权）、delivery-mode（指出该消息可能需要持久性存储）等。生产者把消息交由 RabbitMQ 后，RabbitMQ 会根据消息头把消息发送给感兴趣的 Consumer(消费者)。

2.2 Broker（消息中间件的服务节点）

对于 RabbitMQ 来说，一个 RabbitMQ Broker 可以简单地看作一个 RabbitMQ 服务节点，或者RabbitMQ服务实例。大多数情况下也可以将一个 RabbitMQ Broker 看作一台 RabbitMQ 服务器。
Broker 包括两个部分：Exchange 和 Queue

2.3 Exchange(交换器)

在 RabbitMQ 中，消息并不是直接被投递到 Queue(消息队列) 中的，中间还必须经过 Exchange(交换器) 这一层，Exchange(交换器) 会把我们的消息分配到对应的 Queue(消息队列) 中。

RabbitMQ 的 Exchange(交换器) 有4种类型，不同的类型对应着不同的路由策略：direct(默认)，fanout, topic, 和 headers，不同类型的Exchange转发消息的策略有所区别。后面会在交换器类型介绍到。

生产者将消息发给交换器的时候，一般会指定一个 RoutingKey(路由键) ，用来指定这个消息的路由规则，而这个 RoutingKey 需要与交换器类型和绑定键(BindingKey)联合使用才能最终生效。

RabbitMQ 中通过 Binding(绑定) 将 Exchange(交换器) 与** Queue(消息队列)** 关联起来，在绑定的时候一般会指定一个 BindingKey(绑定建) ,这样 RabbitMQ 就知道如何正确将消息路由到队列了,如下图所示。

2.4 Queue(消息队列)

Queue(消息队列) 用来保存消息直到发送给消费者。它是消息的容器，也是消息的终点。一个消息可投入一个或多个队列。消息一直在队列里面，等待消费者连接到这个队列将其取走。

多个消费者可以订阅同一个队列，这时队列中的消息会被平均分摊（Round-Robin，即轮询）给多个消费者进行处理，而不是每个消费者都收到所有的消息并处理，这样避免消息被重复消费。

3、工作流程

生产者发送消息流程：
1、生产者和Broker建立TCP连接。
2、生产者和Broker建立通道。
3、生产者通过通道消息发送给Broker，由Exchange将消息进行转发
4、Exchange将消息转发到指定的Queue（队列）
消费者接收消息流程：
1、消费者和Broker建立TCP连接
2、消费者和Broker建立通道
3、消费者监听指定的Queue（队列）
4、当有消息到达Queue时Broker默认将消息推送给消费者。
5、消费者接收到消息。
6、ack回复

四、RabiitMQ的五种队列

1、 Exchange Types(交换器类型)

RabbitMQ 常用的 Exchange Type 有 fanout、direct、topic、headers 这四种（AMQP规范里还提到两种 Exchange Type，分别为 system 与自定义）。

1.1 fanout

fanout 类型的Exchange路由规则非常简单，它会把所有发送到该 Exchange 的消息路由到所有与它绑定的 Queue 中，不需要做任何判断操作，所以 fanout 类型是所有的交换机类型里面速度最快的。

fanout 类型常用来广播消息。

1.2 direct

direct 类型的Exchange路由规则也很简单，它会把消息路由到那些 Bindingkey 与 RoutingKey 完全匹配的 Queue 中。

direct 类型常用在处理有优先级的任务，根据任务的优先级把消息发送到对应的队列，这样可以指派更多的资源去处理高优先级的队列。

1.3 topic

topic类型的交换器在匹配规则上进行了扩展，它与 direct 类型的交换器相似，也是将消息路由到 BindingKey 和 RoutingKey 相匹配的队列中，但这里的匹配规则有些不同。

RoutingKey 为一个点号“．”分隔的字符串（被点号“．”分隔开的每一段独立的字符串称为一个单词），如 “com.rabbitmq.client”、“java.util.concurrent”、“com.hidden.client”;
BindingKey 和 RoutingKey 一样也是点号“．”分隔的字符串；
BindingKey 中可以存在两种特殊字符串“”和“#”，用于做模糊匹配，其中“”用于匹配一个单词，“#”用于匹配多个单词(可以是零个)。

1.4 headers

headers 类型的交换器不依赖于路由键的匹配规则来路由消息，而是根据发送的消息内容中的 headers 属性进行匹配。在绑定队列和交换器时制定一组键值对，当发送消息到交换器时，RabbitMQ会获取到该消息的 headers（也是一个键值对的形式) 对比其中的键值对是否完全匹配，如果完全匹配则消息会路由到该队列，否则不会路由到该队列。

headers 类型的交换器性能会很差，而且也不实用，基本上不会看到它的存在。

2、简单队列模式

一个生产者对应一个消费者
生产者将消息发送到“hello”队列。消费者从该队列接收消息。

上面的基本案例就这个模式

3、work模式

竞争消费者模式。
一个生产者对应多个消费者，但是只能有一个消费者获得消息

4、发布/订阅模式

一个消费者将消息首先发送到交换器，交换器绑定到多个队列，然后被监听该队列的消费者所接收并消费。
声明Exchange，不再声明Queue；发送消息到Exchange，不再发送到Queue

5、路由模式

生产者发送的消息会指定一个路由key，那么消息只会发送到相应key相同的队列，接着监听该队列的消费者消费消息。
让消费者有选择性的接收消息。
P：生产者，向Exchange发送消息，发送消息时，会指定一个routing key。
X：Exchange（交换机），接收生产者的消息，然后把消息递交给与routing key完全匹配的队列
C1：消费者，其所在队列指定了需要routing key 为 error 的消息
C2：消费者，其所在队列指定了需要routing key 为 info、error、warning 的消息

6、主题(通配符)模式

每个消费者监听自己的队列，并且设置带统配符的routingkey,生产者将消息发给broker，由交换机根据routingkey来转发消息到指定的队列。
上面的路由模式是根据路由key进行完整的匹配（完全相等才发送消息），这里的通配符模式通俗的来讲就是模糊匹配。
符号“#”表示匹配一个或多个词，符号“*”表示匹配一个词。

例子链接：
https://blog.csdn.net/kavito/article/details/91403659

五、SpringBoot整合RabbitMQ使用

引用文章：
https://blog.csdn.net/kavito/article/details/91403659
http://events.jianshu.io/p/cee2ed448c04
https://www.cnblogs.com/codeli/p/13939632.html

标签：交换器,入门,Exchange,队列,RabbitMQ,Queue,消息
From： https://www.cnblogs.com/Snowclod/p/16914773.html