RabbitMQ(一)---入门

2019-08-15  本文已影响0人  Coding小聪

1. 概述

RabbitMQ是四大热门消息队列中间件（其他的分别是ActiveMQ/RocketMQ/ Kafka）之一，另外它严格遵循AMQP协议。消息队列简单点说就是一个生产者和消费者模型，主要负责接收、存储和投递消息。

2. 安装

这里讲解一下如何安装单机版的RabbitMQ，学习或者非生产环境使用。

2.1. Mac安装

打开终端，直接执行以下指令

brew install rabbitmq

通过以下指令启动RabbitMQ

brew services start rabbitmq

启动完成之后，打开浏览器访问：http://localhost:15672

2.2. Windows安装

安装RabbitMQ之前，需要先安装Erlang。

  1. 先去官网下载Erlang的Windows的安装包，安装的过程就是一路next;
  2. 配置环境变量ERLANG_HOME

  1. 下载rabbitmq的windows安装包，一路next；
  2. 打开cmd，进入rabbitmq的安装目录下的sbin目录，执行rabbitmq-plugins enable rabbitmq_management，启动rabbitmq的管理插件，包括web控台；
  3. 浏览器进入http://localhost:15672/，默认用户名和密码:guest/guest，进入之后可以看到以下页面：

3. Hello World程序

3.1. 生产者

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.util.concurrent.TimeoutException;

public class Send {
    private final static String QUEUE_NAME = "hello.august";

    public static void main(String[] argv)
            throws java.io.IOException, TimeoutException {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("127.0.0.1");

        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare(QUEUE_NAME,"fanout");
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        channel.queueBind(QUEUE_NAME,QUEUE_NAME,"");
        String message = "hello 333";
        channel.basicPublish(QUEUE_NAME, QUEUE_NAME, null, message.getBytes());
        System.out.println("send message: " + message);

        channel.close();
        connection.close();
    }
}

生产者发送消息的流程

  1. 生产者连接到Broker，建立一个连接（Connection），开启一个信道(Channel)；
  2. 生产者声明一个交换器，并设置相关属性，比如交换器类型，是否持久化等；
  3. 生产者声明一个队列并设置相关属性；
  4. 生产者通过路由键将交换器和队列绑定起来；
  5. 生产者发送消息到RabbitMQ Broker，其中包含路由键、交换器等信息；
  6. 相应的交换器根据接收到消息的路由键查找相匹配的队列，如果找到则投递到相应的队列中，如果没有找到，则根据生产者配置的属性选择丢弃还是回退给生产者；
  7. 关闭信道；
  8. 关闭连接。

3.2. 消费者

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.Random;
import java.util.concurrent.TimeoutException;

public class Recv {
    private final static String QUEUE_NAME = "hello.august";

    public static void main(String[] args) {
        try {
            consumer("consumer1");
            //consumer("consumer2");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static void consumer(final String flag) throws IOException, TimeoutException {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("127.0.0.1");
        Connection connection = connectionFactory.newConnection();

        final Channel channel = connection.createChannel();
        channel.exchangeDeclare(QUEUE_NAME,"fanout");
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        channel.queueBind(QUEUE_NAME,QUEUE_NAME,"");
        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body)
                    throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println("["+flag+"] Received '" + message + "'");

                channel.basicReject(envelope.getDeliveryTag(),true);
            }
        };
        channel.basicConsume(QUEUE_NAME, false, consumer);
    }
}

消费者消费消息的流程

  1. 消费者连接到RabbitMQ Broker，并建立一个连接(Connection)，开启一个信道(Channel)；
  2. 消费者向RabbitMQ Broker请求消费相应队列中的消息，可能会做一些队列声明、绑定关系的准备工作，以及设置相应的回调方法；
  3. 等待RabbitMQ Broker回调投递消息，消费者接收消息进行处理；
  4. 消费者向RabbitMQ Broker确认(ack)接收到的消息；
  5. RabbitMQ从队列中删除相应的已经被确认的消息；
  6. 关闭信道；
  7. 关闭连接。

4. 相关概念介绍

RabbitMQ当中涉及以下概念：

4.1 消息（Message）

消息可以近似地看成是业务逻辑数据。之所以说近似，是因为消息一般包含2个部分：消息体（payload）和标签（Label），而业务逻辑数据只存放在消息体当中。消息的标签用来描述这条消息附加值，例如一个交换器的名称或一个路由键，RabbitMQ会根据标签把消息投递给感兴趣的消费者。

在消息路由的过程中，消息的标签会被丢弃，当消息到达队列中只会有消息体。即消费者只会消费到消息体。

4.2 生产者（Producer）

生产者就是生产消息，并将消息投递到RabbitMQ服务器中的一方。

4.3 消费者（Consumer）

消费者就是接收并处理消息的一方。多个消费者可以订阅同一个队列，默认采用轮询的方式进行处理。但是RabbitMQ默认不支持队列层面的消息广播。

4.4 Broker

消息中间件的服务节点，大多数情况下也可以将一个RabbitMQ Broker看作一台RabbitMQ服务器。

4.5 队列（Queue）

队列用于存储消息，RabbitMQ中消息只能存储在队列中。RabbitMQ的生产者生产消息并最终投递到队列中，消费者从队列中获取消息并消费。

4.6 交换器（Exchange）

通常情况下，RabbitMQ的生产者并不会直接将消息发送到队列中，而是先发送到交换器，然后由交换器路由到一个或多个队列中。如果交换器路由不到，或许可以返回给生产者，或许直接丢弃。

4.7 路由键（RoutingKey）

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

在交换器类型和绑定键固定的情况下，生产者在发送消息给交换器时，可以通过指定RoutingKey来决定消息流向哪个队列。

4.8 绑定建（BindingKey）

将交换器和队列关联起来的键，称之为绑定键。

4.9 绑定（Binding）

RabbitMQ中通过绑定将交换器与队列关联起来，在绑定的时候一般会指定一个绑定键，这样RabbitMQ就知道如何正确将消息路由到队列。

4.10 连接（Connection）

无论是生产者还是消费者，都需要和RabbitMQ Broker建立连接，这个连接就是一条TCP连接，也就是Connection。

4.11 通道（Channel）

创建完Connection之后，还需要创建Channel。RabbiMQ处理的每条AMQP指令都是通过Channel完成的。

Channel的出现是为了复用Connection的TCP连接，我们知道创建和销毁Tcp连接都比较耗费资源

Connection可以用来创建多个Channel实例，但是Channel实例不能在线程间共享，应用程序应该为每个线程开辟一个Channel。

5. 交换器类型

RabbitMQ常用的交换器类型有fanout、direct、topic和headers四种。除了这四种之外，AMQP协议还支持两种类型：System和自定义，不过RabbitMQ没有对其进行支持。

它会把所有发送到该交换器的消息路由到所有与此交换器绑定的队列中


fanout

direct类型的交换器会把消息路由到那些BindingKey和RoutingKey完全匹配队列中。

direct

按照图示，如果在发送消息的时候设置路由键为“black”或者“green”，消息会路由到队列Q2。如果设置消息的路由键为“orange”，则会路由到队列Q1。否则，消息会被丢弃或者返回给生产者

topic类型的交换器在direct类型之上做了升级，direct要求BindingKey和RoutingKey等值匹配，而topic只需BindingKey和RoutingKey模糊匹配即可。具体规则如下：

  1. BindingKey和RoutingKey均是一个点号“.”分割的字符串。如，“cn.zgc.rabbitmq”；
  2. BindingKey中允许两种特殊的字符“*”和“#”，其中星号“*”用于匹配一个单词，“#”用于匹配0或多个单词。
topic

以上图的配置为例：
a. 路由键为“lazy.orange.rabbit”的消息会同时路由到Q1和Q2；
b. 路由键为“lazy.q1”的消息只会路由到Q2；
c. 路由键为“cn.orange.red”的消息只会路由到Q1；
d. 路由键为“cn.zgc”的消息由于没有匹配的信息，将会被丢弃或者返回给生产者。

headers类型的交换器根据发送的消息内容中的headers属性来进行匹配。在绑定队列和交换器时指定一组键值对，当Exchange接收到消息时，会获取该消息的headers，然后和绑定时的比较。

headers类型的交换器性能比较差且不实用，所以很少应用于实际开发中。

上一篇下一篇

猜你喜欢

热点阅读