中间件之RabbitMQ

RabbitMQ

1. RabbitMQ简单介绍

RabbitMQ是一个实现了AMQP（Advanced Message Queuing Protocol）高级消息队列协议的消息队列服务，使用以高性能、健壮以及可伸缩性出名的Erlang语言编写的。

主要有两个组件：Exchange和 Queue （在 AMQP 1.0 里还会有变动），如下图所示，绿色的 X 就是 Exchange ，红色的是 Queue ，这两者都在 Server 端，又称作 Broker ，这部分是 RabbitMQ 实现的，而蓝色的则是客户端，通常有 Producer 和 Consumer 两种类型：

接下来简单介绍一下RabbitMQ的工作原理：

• Broker：消息队列服务进程，此进程包括两个部分：Exchange 和 Queue
  • Exchange：消息队列交换机，按一定的规则将消息路由转发到某个队列，对消息进行过滤
  • Queue：消息队列，存储消息的队列，消息到达队列并转发给指定的消费方
• Producer：消息生产者，即生产方客户端，生产方客户端将消费发送到 MessageQueue
• Consumer：消息消费者，即消费方客户端，接收 MessageQueue 转发的消息
• Channel：传输消息的通道，通过通道Producer和Consumer可以与Broker进行通信
• Message：生产者发送的数据，一般有两部分：
  • payload：有效载荷，即传输的数据
  • label：标签，它指的是Exchange的名字或者是一个tag，用来描述了payload

上面的解释有点太专业了，我们现在在此基础上来解释这些名词，并补充一些说法：

Producer

一般一个Message有两个部分：payload 和 label。RabbitMQ也是通过这个label来决定把这个Message发给对应的Consumer。

Consumer

如果有多个Consumer同时订阅同一个Queue中的消息，Queue中的消息会被平摊给多个消费者

Exchange

Exchange是RabbitMQ内的消息交换器。一个Exchange从Producer接收到消息后，一般会指定一个Routing Key，来指定这个消息的路由规则。另外，Routing Key 需要与 Exchange Type 和 Binding Key 联合使用才能最终生效：根据路由规则，匹配查询表中的Routing Key，分发消息到指定的Queue。

Exchange Type

交换机Exchange有如下几种工作模式：

• Direct：点对点发送模式
• Topic：发布订阅模式
• Fanout：分发模式

Queue

队列Queue是RabbitMQ的内部对象，用于存储消息Message。Message最终被Exchange送到Queue中等待Consumer消费，一个Message可以同时被拷贝到多个Queue中。

其实Consumer为了接受Producer的消息，它就得先跟指定Queue进行绑定，绑定后等待Producer将消息传进来，而Queue的创建就是在Consumer要进行绑定前，Producer和Consumer协商好之后，由RabbitMQ创建的。

• 如果Producer发送的消息没有传到对应的Queue的话，那么既Consumer不会得到相应的Message，又Message就会被丢弃。
• 为了使数据不丢失，也就是要有接收数据的Queue，Consumer和Producer都要尽量创建一个Queue（双方商量好要怎么接收数据）。
• 如果Queue绑定了多个Consumer，RabbitMQ使用循环的方式均衡地发送给不同的Consumer。

Binding

在绑定Exchange和Queue的同时，一般会指定一个BindingKey。但是BindingKey并不是在所有情况下都生效，它依赖于ExchangeType。比如在只有介绍的Fanout模式下，在Exchange接收到消息后，就不会将RoutingKey和BindingKey进行比较。

Broker

Broker也就是RabbitMQ Server，作用是维护一条从Producer到Consumer的消息传输路线，保证数据能够按照指定的方式进行传输。

Connection & Channel

Connection 和 Channel都是RabbitMQ对外提供API中最基本的对象。

A Connection represents a real TCP connection to the message broker, whereas a Channel is a virtual connection (AMPQ connection) inside it.

Connection 就是一个连接RabbitMQ Server（即Broker）的TCP连接，Producer和Consumer都是通过TCP连接到RabbitMQ Server的。
Channel是建立在Connection中的虚拟连接，由于建立TCP Connection的开销是巨大的，所以使用Channel来节省开销。Channel是Consumer和Producer对RabbitMQ Server进行一些操作的重要接口，它们大部门业务逻辑操作都是在Channel这个接口中完成的，包括：

• 定义Queue：channel.queueDeclare()
• 定义Exchange：channel.exchangeDeclare()
• 绑定Queue：channel.queueBind()
• 绑定Exchange：channel.exchangeBind()
• 发布消息：
• 接收消息：

之后将单独来介绍Java API！

Virtual Host

Virtual Host是虚拟主机，当有多个不同的用户使用同一个RabbitMQ Server提供的服务时，可以划分出多个VHost，每个用户在自己的VHost中创建Exchange和Queue。

待展开

小节总结

生产者在发送Message时，需要指定Exchange和RoutingKey，Exchange收到Message后，根据自身的ExchangeType按照一定规则将消息转发到相应的Queue中。

• Exchange Type为Direct时

  • Exchange通过Message中的RoutingKey，与Binding中的BindingKey进行比较，若匹配，则将消息发送到对应的Queue中。
    模拟AProducer将带有RoutingKey为toAConsumer的消息通过AExchange发送到AConsumer：
    
    这里Producer发送的Message情况如下：
    
    最后只有BindingKey为toAConsumer的消息队列toAConsumer接收到了消息，所以符合点对点发布。

• Exchange Type为Topic时

  • Exchange通过Message中的RoutingKey，与Binding中的BindingKey进行比较，若匹配，则将消息发送到对应的Queue中。
  • Topic Exchange说白了就是Direct Exchange的通配符模式
  • 这里RoutinKey和BindingKey应该设置为.*.*.*.*这种形式
    

• Exchange Type为Fanout时

  • Exchange直接将Message发送到与Exchange绑定的Queue，然后Message都能被绑定的Queue接收到，就像广播一样。
    可见AProducer发送的所有消息都能被AExchange发送给两个与之绑定的队列，然后被两个Consumer消费。

2. RabbitMQ安装及使用

RabbitMQ的下载：

在使用RabbitMQ之前需要安装RabiitMQ的运行环境Erlang，就好比使用ActiveMQ的时候需要依赖Java环境，Erlang下载。

接下来运行RabbitMQ：

>cd F:\ProgramFiles\RabbitMQServer\rabbitmq_server-3.8.7\sbin
>.\rabbitmq-plugins enable rabbitmq_management

>.\rabbitmqctl start_app
>.\rabbitmqctl status

这边要注意：

• RabbitMQ的安装路径不能出现中文或者空格！
• 如果运行>.\rabbitmqctl start_app时出现(ARGUMENTERROR) ARGUMENT ERROR的话就要将系统用户名改成英文！
  • https://blog.csdn.net/wls666/article/details/103334152按照上面的修改用户名后，重新运行上面的代码就可以运行了！
  • https://blog.csdn.net/wls666/article/details/103326096参考这个博。
    
    解决办法

出错总结：

• 安装路径可能出现中文或者空格
• 系统用户名可能出现中文，要改成英文
• 某个端口被其他占用了（这边我直接将ActiveMQ的服务关闭就可以正常使用了！）

最终出现如下结果表示说明安装是成功的，并且说明现在RabbitMQ Server已经启动了，运行正常，于是就可以打开http://localhost:15672

登录密码都是guest，登录之后可以自己创建管理员：

安装也可以参考这个博主
如果没有administrator的话，就会提示Not management user：

>.\rabbitmqctl list_users
> .\rabbitmqctl set_user_tags guest administrator

3. RabbitMQ的常用操作

待补充

4. RabbitMQ的JavaAPI

5. RabbitMQ Demo for Java

现在我们有若干台试验设备（使用C/C++编程，但是这边假设使用Java），做实验的学生一人连接着一台实验设备，如果每个学生要将实验设备中的数据传输到学生电脑中，我们应该：

1. 学生端向实验设备发送请求数据消息
2. 实验设备向学生端发送实验数据
   有了以上两点分步，我们就可以很明确每步实验设备和学生的角色了，还有就是这两步都是采用发布订阅模式（Pub/Sub）进行Messaging的：

学生向实验设备发送请求数据消息

此时，学生（Producer）向某个实验设备（Consumer）发送请求数据的消息，这边Exchange Type为Topic，RoutingKey和BindingKey我们设置为send.request.*的形式，

实验设备向学生发送实验数据

反过来，实验设备（Producer）收到学生（Consumer）的请求后，将实验数据传输到学生的电脑上：

Demo代码

首先创建一个工厂类，来返回Connection：

package com.shallwe.rabbitmq.util;

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

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

/**
 * 连接配置
 */
public class ConnectionUtils {
    private static String HOST="localhost";
    private static Integer PORT=5672;
    private static String VIRTUAL_HOST="ShallWe";
    private static String USER_NAME="admin";
    private static String PASSWORD="admin";

    /**
     * 获取RabbitMQ的连接
     * @return
     * @throws IOException
     * @throws TimeoutException
     */
    public static Connection getConnection() throws IOException, TimeoutException {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        //设置服务地址
        connectionFactory.setHost(HOST);
        //设置端口号
        connectionFactory.setPort(PORT);
        //设置virtual hosts
        connectionFactory.setVirtualHost(VIRTUAL_HOST);
        //设置用户名
        connectionFactory.setUsername(USER_NAME);
        //设置密码
        connectionFactory.setPassword(PASSWORD);
        //从工厂返回一个连接
        return connectionFactory.newConnection();
    }


}

实验设备的代码：

package com.shallwe.rabbitmq.demo.labfacility;

import com.rabbitmq.client.*;
import com.shallwe.rabbitmq.util.ConnectionUtils;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.lang.reflect.Method;
import java.text.ParseException;

public class LabFacility1 {
    private static String EndName = "LabFacility1";
    private static String toWhom;
    private static String receiveExchange = "AExchangeForTopic";
    private static String sendExchange = "BExchangeForTopic";

    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtils.getConnection();
        String queueName = "to" + EndName;
        Channel channel = connection.createChannel();
        channel.exchangeDeclare(receiveExchange, "topic");      //接收请求消息Exchange
        channel.exchangeDeclare(sendExchange, "topic");      //发送实验数据Exchange
        channel.queueDeclare(queueName, false, false, false, null);
        channel.queueBind(queueName, receiveExchange, "send.request.1");        //后面这个bindingKey很关键
        //实验设备每次先接收到请求消息后再向学生发送实验数据
        System.out.println("监听是否有请求消息传入....");
        DefaultConsumer LabFacility = 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("[Facility Receive]:" + message);
                if (true) {
                    //学生端发送的消息是数字，那么就发送实验数据
                    System.out.println("实验设备"+EndName+"即向学生端Student"+message+"发送数据....");
                    for (int i = 1; i <= 10; i++) {
                        String labData = "实验数据" + i;
                        //这边队名使用Consumer的名称
                        channel.basicPublish(sendExchange, "send.labdata."+message,
                                null, labData.getBytes("utf-8"));
                    }
                } else {
                    System.out.println("学生端向实验设备发送其他请求，功能待开发！");
                }
                try {
                    Thread.sleep(1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    System.out.println("Waitting to send Lab data!");
                    //手动应答
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        };
        Boolean autoAck = false;
        channel.basicConsume(queueName, autoAck, LabFacility);
    }
}

学生端的代码：

package com.shallwe.rabbitmq.demo.student;

import com.rabbitmq.client.*;
import com.shallwe.rabbitmq.util.ConnectionUtils;


import java.io.IOException;
import java.lang.reflect.Method;
import java.util.Arrays;
import java.util.List;

public class Student1 {
    private static String EndName="Student1";
    private static String toWhom;
    private static String sendExchange="AExchangeForTopic";
    private static String receiveExchange="BExchangeForTopic";

    public static void main(String[] args) throws Exception{
        Connection connection = ConnectionUtils.getConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare(sendExchange, "topic");
        channel.exchangeDeclare(receiveExchange, "topic");
        String queueName="to"+EndName;
        channel.queueDeclare(queueName,false, false, false, null);
        channel.queueBind(queueName, receiveExchange, "send.labdata.1");
        DefaultConsumer student=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("["+EndName+" Receives]:"+message);
                try{
                    Thread.sleep(1000);
                }catch(InterruptedException e){
                    e.printStackTrace();
                }finally{
                    //最后要进行手动应答
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        };
        //模拟：Student1要求LabFacility2发送数据，发个2就行了！
        channel.basicPublish(sendExchange, "send.request.2", null, "2".getBytes("utf-8"));
        //模拟：Student1接收到LabFacility发来的数据
        Boolean autoAck=false;
        channel.basicConsume(queueName,autoAck,student);
        //接下来可以试一下客户端发送信息
    }
}