《分布式消息中间件实践》 读书笔记

Posted 2021-08-04 山河已无恙

---

写在前面

---

• 听人家总说这些消息中间件，之前项目里也没有用到，其实挺好奇的,之前也没时间看。这里准备用这本书入门，查了一下，书评很一般，但是入门应该够了，而且200多页，正合适，生活加油
• 笔记主要是书的摘写，不懂的地方写一些自己的理解。
• 嗯，看了一些发现东西挺多，是我唐突了…,不简单。
• 笔记还在更新中。

夫人之相与，俯仰一世，或取诸怀抱，悟言一室之内；或因寄所托，放浪形骸之外--------《兰亭集序》东晋·王羲之

---

第1章消息队列

1.1 系统间通信技术介绍

• 如果是一个业务被拆分成多个子业务部署在不同的服务器上,那就是分布式应用;
• 如果是同一个业务部署在多台服务器上,那就是集群。

分布式应用的子系统之间并不是完全独立的,它们需要相互通信来共同完成某个功能,这就涉及系统间通信了。

目前,业界通常有两种方式来实现系统间通信,

• 基于远程过程调用的方式(RPC调用);
• 基于消息队列的方式。

RPC是一种通过网络从远程计算机程序上请求服务,而不需要了解底层网络技术的协议。

• 它是协议,是一种规范,就需要有遵循这套规范的实现。典型的RPC实现包括Dubbo、Thrift、 GRPC等。
• 网络通信的实现是透明的,调用方不需要关心网络之间的通信协议、网络IO模型、通信的信息格式等。
• 跨语言,调用方实际上并不清楚对端服务器使用的是什么程序语言。对于调用方来说,无论其使用的是何种程序语言,调用都应该成功,并且返回值也应按照调用方程序语言能理解的形式来描述。
  

基于消息队列的方式是指由应用中的某个系统负责发送信息,由关心这条消息的相应系统负责接收消息,并在收到消息后进行各自系统内的业务处理。消息可以非常简单,比如只包含文本字符串;也可以很复杂,比如包含字节流、字节数组,还可能包含嵌入对象,甚至是Java·对象(经过序列化的对象)。

1.2 为何要使用消息队列

消息队列的典型场景-异步处理，解耦、流量削峰、日志收集、事务最终一致性等问题。

• 异步处理：这个可以结合AJAX理解，比如AJAX通过异步请求，这个异步请求是可以随时发出的，那么在Tomcat里就有相应的工作队列来存储请求信息。
• 模块解耦:随着需求的叠加,各模块之间逐渐变成了相互调用的关系,这种模块间紧密关联的关系就是紧耦合。紧耦合带来的问题是对一个模块的功能变更将导致其关联模块发生变化,因此各个模块难以独立演化。要解决这个问题,可以在模块之间调用时增加一个中间层来实现解耦,这也方便了以后的扩展。所谓解耦,简单地讲,就是一个模块只关心自己的核心流程,而依赖该模块执行结果的其他模块如果做的不是很重要的事情,有通知即可,无须等待结果。换句话说,基于消息队列的模型,关心的是通知,而非处理。
• 流量削峰 某一时刻网站突然迎来用户请求高峰期的情况,如果在设计上考虑不周甚至会发生雪崩(在分布式系统中,经常会出现某个基础服务不可用造成整个系统不可用的情况,这种现象被称为“服务雪崩效应”),从而发生整个系统不可用的严重生产事故。当访问量剧增时系统依然可以继续使用,该怎么做呢?首先想到的是购买更多的服务器进行扩展,以增强系统处理并发请求的能力。如果都以能处理此类流量峰值为标准投入大量资源随时待命无疑是很大的浪费。在业界的诸多实践中,常见的是使用消息队列,先将短时间高并发的请求持久化,然后逐步处理,从而削平高峰期的并发流量,改善系统的性能
• 日志收集 利用消息队列产品在接收和持久化消息方面的高性能,引入消息队列快速接收日志消息,避免因为写入日志时的某些故障导致业务系统访问阻塞、请求延迟等。所以很多公司会选择构建一个日志收集系统,由它来统一收集业务日志数据,供离线和在线的分析系统使用。
• 事务最终一致性 :业界曾经提出过一个处理分布式事务的规范-XAXA主要定义了全局事务管理器(Transaction Manager)和局部资源管理器(Resource Manager)之间的接口.XA接口是双向的系统接口,在事务管理器及一个或多个资源管理器之间形成通信桥梁。XA引入的事务管理器充当全局事务中的协调者的角色。事务管理器控制着全局事务,管理事务生命周期,并协调资源。资源管理器负责控制和管理实际资源(如数据库或JMS队列)。目前各主流数据库都提供了对XA规范的支持。XA所以它的最大缺陷是性能很差,因此并不适合在生产环境下有高并发和高性能要求的场景。在业界的很多实践方案中,都可以借助消息队列来处理此问题。

1.3 消息队列的功能特点

一个典型意义上的消息队列,至少需要包含消息的发送、接收和暂存功能

在生产环境应用中,对消息队列的要求远不止基本的消息发送、接收和暂存。在不同的业务场景中,需要消息队列产品能解决诸如消息堆积、消息持久化、可靠投递、消息重复、严格有序、集群等各种问题。

1. 消息堆积:消息在处理中心逐渐积压而得不到释放。比如给消息队列设置一个阈值,将超过阈值的消息不再放入处理中心,以防止系统资源被耗尽,导致机器挂掉甚至整个消息队列不可用。
2. 消息持久化: 将消息放在内存中存在的最大问题是,一旦机器宿掉消息将丢失。持久化方案有很多种,比如将消息存到本地文件、分布式文件系统、数据库系统中等。
3. 可靠投递:可靠投递是不允许存在消息丢失的情况的。从消息的整个生命周期来分析,消息丢失的情况一般发生在如下过程中:
   • 从生产者到消息处理中心。
   • 从消息处理中心到消息消费者。
   • 消息处理中心持久化消息。
4. 消息重复:为了支持消息可靠投递,当消息发送失败或者不知道是否发送成功时(比如超时),消息的状态是待发送,定时任务不停地轮询所有的待发送消息,最终保证消息不会丢失,这就带来了消息可能会重复的问题。
5. 严格有序 : 在实际的业务场景中,经常会碰到需要按生产消息时的顺序来消费的情形。需要消息队列能够提供有序消息的保证。但顺序消费却不一定需要消息在整个产品中全局有序,有的产品可能只需要提供局部有序的保证。
6. 集群:系统架构一般都需要实现高可用性,以排除单点故障引起的服务中断,保证7x24小时不间断运行,所以可能需要消息队列产品提供对集群模式的支持。集群不仅可以让消费者和生产者在某个节点崩溃的情况下继续运行,集群之间的多个节点还能够共享负载,当某台机器或网络出现故障时能自动进行负载均衡,而且可以通过增加更多的节点来提高消息通信的吞吐量。
7. 消息中间件:消息中间件关注于数据的发送和接收,利用高效、可靠的异步消息传递机制集成分布式系统。消息传输中间件(MOM)简化了应用之间数据的传输,屏蔽了底层的异构操作系统和网络平台,提供了一致的通信和应用开发标准,确保在分布式计算网络环境下可靠、跨平台的信息传输和数据交换。它基于消息队列的存储-转发机制,并提供了特有的异步传输机制,能够基于消息传输和异步事务处理实现应用整合与数据交换。

中间件:非底层操作系统软件、非业务应用软件,不是直接给最终用户使用的,不能直接给客户带来价值的软件统称为中间件。

1.4 设计一个简单的消息中间件

1.4.1 消息处理中心

实现一个消息队列。代码详见https://gitee.com/liruilonger/workspack/tree/master/src/com/msg_queue/jkd

package com.msg_queue.jkd;

import java.util.concurrent.ArrayBlockingQueue;

/**
 * @Classname Broker
 * @Description TODO 消息处理中心类（ Broker ）
 * @Date 2021/7/5 0:37
 * @Created Li Ruilong
 */
public class Broker {

    ///队列存储消息的最大数量
    private  final  static int MAX_SIZE = 3;

    //保存消息数据的容
    private  static ArrayBlockingQueue<String> messageQueue = new ArrayBlockingQueue<>(MAX_SIZE);

   // 生产消息
   public static void  produce(String msg){
        if (messageQueue.offer(msg)){
            System.out.printf("投递成功%s,当前暂存消息数量%d",msg,messageQueue.size());
        }else {
            System.out.println("消息处理中心内暂存的消息达到最大负荷，不能放入消息");
        }

   }

   //消费消息
    public static String consume(){
       //
       String msg = messageQueue.poll();
       if (msg != null){
           System.out.printf("已经消费消息 %s，当前暂存消息数量 %s",msg,messageQueue.size());
       }else {
           System.out.println("消息处理中心没有消息可供消费");
       }
       return msg;
    }
}

BrokerServer用来对提供Broker类的服务。

/**
 * @Classname BrokerServer
 * @Description TODO  定义了BrokerServer类用来对外提供Broker类的服务。
 * @Date 2021/7/5 12:32
 * @Created Li Ruilong
 */
public class BrokerServer  implements Runnable{

    public static  int SERVICE_PORT = 9999;

    private final Socket socket;

    //该Socket是由客户端请求的得到的Socket实例。
    public BrokerServer(Socket socket) {
        this.socket = socket;
    }


    @Override
    public void run() {
        try(
                // 拿到输入流
                BufferedReader in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
                // 拿到输出流
                PrintWriter out = new PrintWriter(socket.getOutputStream())
        ){
            while (true){
                //  拿到输入的数据
                String str = in.readLine();
                if (str == null){
                    continue;
                }
                System.out.println("接受原始消息"+str);
                // CONSUME 表示消费一条消息
                if ("CONSUME".equals(str)){
                    // 消费消息
                    String s = Broker.consume();
                    out.println(s);
                    out.flush();
                }else {
                    // 其他情况表示生产消息放到消息队列里面
                    Broker.produce(str);
                }
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws Exception {
        // 服务端套接字，监听9999 端口
        ServerSocket serverSocket = new ServerSocket(SERVICE_PORT);
        while (true){
            // todo serverSocket.accept()接受客户端Socket连接请求，并返回一个与客户端Socket对应的Socket实例，该方法是一个阻塞方法，
            // 如果没有接受到客户端发送的Socket，则一直处于等待状态，线程也会被阻塞。
            BrokerServer server = new BrokerServer(serverSocket.accept());
            new Thread(server).start();
        }
    }
}

1.4.2 客户端访问

/**
 * @Classname MqClient
 * @Description TODO 客户端
 * @Date 2021/7/5 13:36
 * @Created Li Ruilong
 */
public class MqClient {


    /*
     * @return
     * @Description 消息生产者
     * @author Liruilong
     * @date  2021/7/5  14:06
     **/
    public static  void produce(String message) throws Exception {
        // 利用Socket模拟发送消息的一方。
        Socket socket = new Socket(InetAddress.getLocalHost(),BrokerServer.SERVICE_PORT);
        try (
            PrintWriter out = new PrintWriter(socket.getOutputStream())
        ){
            out.println(message);
            out.flush();
        }

    }


    /*
     * @return
     * @Description 消息消费者
     * @author Liruilong
     * @date  2021/7/5  15:20
     **/
    public static String consume() throws Exception{
         // 利用Socket模拟消费消息的一方。
        Socket socket = new Socket(InetAddress.getLocalHost(),BrokerServer.SERVICE_PORT);
        try (
                BufferedReader in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
                PrintWriter out = new PrintWriter(socket.getOutputStream());
        ){
            out.println("CONSUME");
            out.flush();
            String message = in.readLine();

            return  message;
        }
    }

}

生产消息

/**
 * @Classname ProduceClient
 * @Description TODO   生产消息
 * @Date 2021/7/5 14:36
 * @Created Li Ruilong
 */
public class ProduceClient {

    public static void main(String[] args) throws Exception {
        MqClient client = new MqClient();
        client.produce("Hello Word");

    }
}

消费消息

/**
 * @Classname ConsumeClient
 * @Description TODO 消费消息
 * @Date 2021/7/5 14:39
 * @Created Li Ruilong
 */
public class ConsumeClient {
    public static void main(String[] args) throws Exception {
     MqClient mqClient = new MqClient();
        String consume = mqClient.consume();
        System.out.println("获取的消息为："+consume);
    }
}

第2章消息协议

类似于上免得CONSUME,用与区分连接Socket的是生产消息的客户端还是消费消息的客户端。

消息协议则是指用于实现消息队列功能时所涉及的协议。按照是否向行业开放消息规范文档,可以将消息协议分为开放协议和私有协议。

常见协议有AMOP, MQTT, STOMP,XMPP等。有些特殊框架(如Redis, Kafka, ZeroMQ)根据自身需要未严格遵循MQ规范,而是基于TCP/IP自行封装了一套协议,通过网络Socket接口进行传输,实现了MQ的功能。

这里的协议可以简单地理解成对双方通信的一个约定.

2.1 AMQP

在2004年,摩根大通和iMatrix开始着手Advanced Message Queuing Protocol (AMQP)开放标准的开发。2006年,发布了AMQP规范。目前AMQP协议的版本为1.0。

一般来说,将AMQP协议的内容分为三部分:基本概念、功能命令和传输层协议。

• 基本概念是指AMQP内部定义的各组件及组件的功能说明。
• 功能命令是指该协议所定义的一系列命令,应用程序可以基于这些命令来实现相应的功能。
• 传输层协议是一个网络级协议,它定义了数据的传输格式,消息队列的客户端可以基于这个协议与消息代理和AMQP的相关模型进行交互通信,该协议的内容包括数据帧处理、信道复用、内容编码、心跳检测、数据表示和错误处理等。

1. 主要概念

• Message (消息):消息服务器所处理数据的原子单元。消息可以携带内容,从格式上看,消息包括一个内容头、一组属性和一个内容体。这里所说的消息可以对应到许多不同应用程序的实体,比如一个应用程序级消息、一个传输文件、一个数据流帧等。消息可以被保存到磁盘上,这样即使发生严重的网络故障、服务器崩溃也可确保投递消息可以有优先级,高优先级的消息会在等待同一个消息队列时在低优先级的消息之前发送,当消息必须被丢弃以确保消息服务器的服务质量时,服务器将会优先丢弃低优先级的消息。消息服务器不能修改所接收到的并将传递给消费者应用程序的消息内容体。消息服务器可以在内容头中添加额外信息,但不能删除或修改现有信息。
• Publisher (消息生产者):也是一个向交换器发布消息的客户端应用程序。
• Exchange (交换器):用来接收消息生产者所发送的消息并将这些消息路由给服务器中的队列。
• Binding (绑定):用于消息队列和交换器之间的关联。一个绑定就是基于路由键将交换器和消息队列连接起来的路由规则,所以可以将交换器理解成一个由绑定构成的路由表（路由控制表）。
• Virtual Host (虚拟主机):它是消息队列以及相关对象的集合,是共享同一个身份验证和加密环境的独立服务器域。每个虚拟主机本质上都是一个mini版的消息服务器,拥有自己的队列、交换器、绑定和权限机制。
• Broker (消息代理):表示消息队列服务器,接受客户端连接,实现AMQP消息队列和路由功能的过程。
• Routing Key (路由规则):虚拟机可用它来确定如何路由一个特定消息。
• Queue (消息队列):用来保存消息直到发送给消费者。它是消息的容器,也是消息的终点。一个消息可被投入一个或多个队列中。消息一直在队列里面,等待消费者连接到这个队列将其取走。
• Connection (连接):可以理解成客户端和消息队列服务器之间的一个TCP连接。
• Channel (信道):仅仅当创建了连接后,若客户端还是不能发送消息,则需要为连接创建一个信道。信道是一条独立的双向数据流通道,它是建立在真实的TCP连接内的虚拟连接,AMQP命令都是通过信道发出去的,不管是发布消息、订阅队列还是接收消息,它们都通过信道完成。一个连接可以包含多个信道,之所以需要信道,是因为TCP连接的建立和释放都是十分昂贵的,如果客户端的每一个线程都需要与消息服务器交互,如果每一个线程都建立了一个TCP连接,则暂且不考虑TCP连接是否浪费,就算操作系统也无法承受每秒建立如此多的TCP连接。
• Consumer (消息消费者):表示一个从消息队列中取得消息的客户端应用程序。

2. 核心组件的生命周期

(1)消息的生命周期一条消息的流转过程通常是这样的:

• Publisher产生一条数据,发送到Broker, Broker中的Exchange可以被理解为一个规则表(Routing Key和Queue的映射关系-Binding), Broker收到消息后根据Routing Key查询投递的目标Queue.
• Consumer向Broker发送订阅消息时会指定自己监听哪个Queue,当有数据到达Queue时Broker会推送数据到Consumer.
  

(2)交换器的生命周期
每台AMQP服务器都预先创建了许多交换器实例,它们在服务器启动时就存在并且不能被销毁。如果你的应用程序有特殊要求,则可以选择自己创建交换器,并在完成工作后进行销毁。
(3)队列的生命周期
这里主要有两种消息队列的生命周期,即持久化消息队列和临时消息队列。持久化消息队列可被多个消费者共享,不管是否有消费者接收,它们都可以独立存在。临时消息队列对某个消费者是私有的,只能绑定到此消费者,当消费者断开连接时,该消息队列将被删除。

3. 功能命令

AMQP协议文本是分层描述的,在不同主版本中划分的层次是有一定区别的。
0-9 版本共分两层: Functional Layer (功能层)和Transport Layer (传输层)

• 功能层定义了一系列命令,这些命令按功能逻辑组合成不同的类(Class),客户端应用可以利用它们来实现自己的业务功能。
• 传输层将功能层所接收的消息传递给服务器经过相应处理后再返回,处理的事情包括信道复用、帧同步、内容编码、心跳检测、数据表示和错误处理等.
  0-10 版本则分为三层: Model Layer (模型层)、Session Layer (会话层)和Transport Layer(传输层)。
• 模型层定义了一套命令,客户端应用利用这些命令来实现业务功能。
• 会话层负责将命令从客户端应用传递给服务器,再将服务器的响应返回给客户端应用,会话层为这个传递过程提供了可靠性、同步机制和错误处理。
• 传输层负责提供帧处理、信道复用、错误检测和数据表示

4. 消息数据格式
所有的消息必须有特定的格式来支持,这部分就是在传输层中定义的。AMQP是二进制协议,协议的不同版本在该部分的描述有所不同。0-9-1版本为例,看一下该版本中的消息格式

所有的消息数据都被组织成各种类型的帧(Frame),帧可以携带协议方法和其他信息,所有帧都有同样的格式,都由一个帧头(header, 7个字节)、任意大小的负载(payload)和一个检测错误的结束帧(frame-end)字节组成。其中:

• 帧头包括一个type字段、一个channel字段和一个size字段;
• 帧负载的格式依赖帧类型(type)
  要读取一个帧需要三步。
• ①读取帧头,检查帧类型和通道(channel).
• ②根据帧类型读取帧负载并进行处理。
• ③读取结束帧字节。
  AMQP定义了如下帧类型。
• type =1, “METHOD”:方法帧;
• type=2, “HEADER”:内容头帧;
• type=3,“BODY”:内容体帧;
• type=4, “HEARTBEAT”:心跳帧通道

编号为0的代表全局连接中的所有帧, 1-65535代表特定通道的帧。size字段是指帧负载的大小,它的数值不包括结束帧字节。AMQP使用结束帧来检测错误客户端和服务器实现引起的错误。

2.2 MQTT

MQTT (Message Queuing Telemetry Transport,消息队列遥测传输)是IBM开发的一个即时通信协议,该协议支持所有平台,几乎可以把所有联网物品和外部连接起来,被用来当作传感器和制动器的通信协议。

目前MQTT协议版本为2014年发布的MQTT 3.1.1,它是一个基于TCP/IP协议、可提供发布/订阅消息模式、十分轻量级的通信协议。除标准版外,还有一个简化版MQTI-SN,它基于非TCP/IP协议(如ZigBee协议),该协议主要为嵌入式设备提供消息通信。

这里主要介绍标准版MQTT 3.1.1,该协议是一个基于客户端-服务器的消息发布/订阅传输协议,其特点是轻量、简单、开放和易于实现。正因为这些特点,使它常应用于很多机器计算能力有限、低带宽、网络不可靠的远程通信应用场景中。

IBM WebSphere, MQ Telemetry, IBM Messagesight. Mosquitto, Eclipse Paho, emqttd Xively.m2m.io, webMethods, Nirvana Messaging, RabbitMQ Apache ActiveMQ, Apache Apollo,Moquette, HiveMQ, Mosca, Litmus Automation Loop. JoramMQ, ThingMQ, VerneMQ.

1. 主要概念

所有基于网络连接的应用都会有客户端(Client)和服务器(Server),而在MQTT协议中使用者有三种身份:发布者(Publisher)、代理(Broker)和订阅者(Subscriber)。其中消息的·发布者和订阅者都是客户端,消息代理是服务器,消息发布者可以同时是订阅者。一条消息的流转过程是这样的:先由消息发布者发布消息到代理服务器,在消息中会包含主题(Topic),之后消息订阅者如果订阅了该主题的消息,将会收到代理服务器推送的消息.(基于观察者模式理解)

下面介绍MQTT协议中的基本组件。

(1)网络连接(Network Connection):网络连接指客户端连接到服务器时所使,用的底层传输协议, 由该连接来负责提供有序的、可靠的、基于字节流的双向传输

(2)应用消息(Application Message):应用消息指通过网络所传输的应用数据,该数据一般包括主题和负载两部分。

(3)主题(Topic):主题相当于应用消息的类型,消息订阅者订阅后,就会收到该主题的消息内容。

(4)负载(Payload):负载指消息订阅者具体接收的内容

(5)客户端(Client):客户端指使用MQTT的程序或设备。客户端总是通过网络连接到服务端,它可以发布应用消息给其他相关的客户端、订阅消息用以请求接收相关的应用消息、取消订阅应用消息、从服务器断开连接等。

(6)服务器(Server): 服务器也是指程序或设备,它作为发送消息的客户端和请求订阅的客户端之间的中介。服务器的功能包括接收来自客户端的网络连接、接收客户端发布的应用消息、处理客户端的订阅和取消订阅的请求、转发应用消息给相应的客户端等。

(7) 会话(Session):客户端与服务器建立连接之后就是一个会话,客户端和服务器之间通过会话来进行状态交互。会话存在于一个网络连接之间,也可能会跨越多个连续的网络连接。会话主要用于客户端和服务器之间的逻辑层面的通信。

(8)订阅(Subscription):订阅一般与一个会话关联,会话可以包含多于一个的订阅。订阅包含一个主题过滤器和一个服务质量(Qos)等级。会话的每个订阅都有一个不同的主题过滤器。

(9)主题名(Topic Name):主题名是附加在消息上的一个标签,该标签与服务器的订阅相匹配,服务器会根据该标签将消息发送给与订阅所匹配的每个客户端。

(10)主题过滤器(Topic Filter): 主题过滤器是订阅中包含的一个表达式,用于表示相关联的一个或多个主题。主题过滤器可以使用通配符。

(11) MQTT控制报文(MQTT Control Packet): MQTT控制报文实际上就是通过网络连接发送的信息数据包。

2·消息数据格式

MQTT协议是通过交换预定义的MQTT控制报文来通信的,·内容由三部分组成
固定报头(Fixed header):存在于所有控制报文中,内容包含控制报文类型、相应的标识位和剩余长度
可变报头(Variable header):存在于部分控制报文中,由固定报头中的控制报文类型决定是否需要可变报头,以及可变报头的具体内容。
消息体(Payload):存在于部分控制报文中,表示客户端接收到的具体内容。

嗯，有些多，先不看额，遇到在深入学习。😃

2.3 STOMP

STOMP (Streaming Text Orientated Messaging Protocol,流文本定向消息协议)是一个简单的文本消息传输协议,它提供了一种可互操作的连接格式,允许客户端与任意消息服务器(Broker)进行交互。在设计STOMP时借鉴了HTTP的一些理念,将简易性、互通性作为其主要设计哲学,这使得STOMP协议的客户端的实现很容易。

主要介绍STOMP 1.2版本协议的相关内容。STOMP被设计成轻量级的协议,使得很容易用其他语言来实现客户端和服务器端,因此它在多种语言和平台上得到广泛应用。目前有很多STOMP消息中间件服务器,如下都是STOMP协议的服务器端实现。

Apache Apollo, Apache ActiveMQ, RabbitMQ HornetQ, Stampy, StompServer.

嗯，有些多，简单了解下，先不看额，遇到在深入学习。😃

2.4 XMPP

XMPP (可扩展通信与表示协议)是一种基于XML的流式即时通信协议,它的特点是将上下文信息等嵌入到用XML表示的结构化数据中,使得人与人之间、人与应用系统之间,以及应用系统之间能即时相互通信。XMPP的基本语法和语义最初主要是由Jabber开放源代码社区于1999年开发的,其基础部分早在2002-2004年就得到了互联网工程任务组(IETF)的批准。

XMPP定义了用于通信网络实体之间的开放协议的规范,其规范说明由一系列作用不同的RFC文档组成, 目前核心规范主要包括RFC 6120, RFC 6121,RFC 7622及RFC 7395中定义的WebSocket绑定。

2.5 JMS

JMS (Java Message Service)即Java消息服务应用程序接口,是Java平台中面向消息中间件的一套规范的Java API接口,用于在两个应用程序之间或分布式系统中发送消息,进行异步通信。这套规范由SUN提出, 目前主要使用的版本有两个:一个是2002年发布的1.1版;yi个是2013年发布的2.0版。

不同于本章上面所介绍的AMQP, MQTT,STOMP,XMPP等协议,JMS并不是消息队列协议的一种,更不是消息队列产品,它是与具体平台无关的API, 目前市面上的绝大多数消息中间件厂商都支持JMS接口规范。换句话说,你可以使用JMS API来连接支持AMQP, STOMP等协议的消息中间件产品(比如ActiveMQ, RabbitMQ等),在这一点上它与Java中的JDBC的作用很像,我们可以用JDBC API来访问具体的数据库产品(比如Oraclemysql等)。

1 体系架构

JMS的作用是提供通用接口保证基于JMS API编写的程序适用于任何一种模型,使得在更换消息队列提供商的情况下应用程序相关代码也不需要做太大的改动。

• (1)点对点模型在点对点(Point to Point)模型中,应用程序由队列(Queue)、发送者(Sender)和接收者(Receiver)组成。每条消息都被发送到一个特定的队列中,接收者从队列中获取消息
  
  队列中一直保留着消息,直到它们被接收或超时。点对点 模型的特点如下
  • 每条消息只有一个接收者,消息一旦被接收就不再保留在消息队列中了。
  • 发送者和接收者之间在时间上没有依赖。也就是说,当消息被发送之后,不管接收者有没有在运行,都不会影响消息被发送到队列中。
• (2)发布/订阅模型在发布/订阅(Pub/Sub)模型中,应用程序由主题(Topic)、发布者(Publisher)和订阅者(Subscriber)组成。发布者发布一条消息,该消息通过主题传递给所有的订阅者
  
  在这种模型中,发布者和订阅者彼此不知道对方,它们是匿名的并且可以动态发布和订阅主题。主题用于保存和传递消息,并且会一直保存消息直到消息被传递给订阅者。发布/订阅模型的特点如下:
  • 每条消息可以有多个订阅者。
  • 发布者和订阅者之间有时间上的依赖。一般情况下,某个主题的订阅者需要在创建了订阅之后才能接收到消息,而且为了接收消息订阅者必须保持运行的状态。
  • JMS允许订阅者创建一个可持久化的订阅,这样即使订阅者没有运行也能接收到所订阅的消息。
  • 每条消息都会传送给该主题下的所有订阅者。通常发布者不会知道也意识不到哪一个订阅者正在接收消息。

2.基本概念

按照JMS规范中所说的,一个JMS应用由如下几个部分组成。

• JMS客户端(JMS Client):指发送和接收消息的Java程序。
• 非JMS客户端(Non-JMS Client):指使用消息系统原生的客户端API代替JMS的客户端。如果应用程序在JMS规范前就已存在,则它可能同时包含JMS客户端和非JMS客户端。
• 消息(Message):每个应用都定义了一组消息,用于多个客户端之间的消息通信。
• JMS提供商(JMS Provider):指实现了JMS API的实际消息系统。
• 受管对象(Administered Object):指由管理员创建,并预先配置好给客户端使用的JMS对象。JMS中的受管对象分为两种,即ConnectionFactory (客户端使用这个对象来创建到提供者的连接)和Destination (客户端使用这个对象来指定发送或接收消息的目的地)。

而具体到JMS应用程序,则主要涉及以下基本概念。

• 生产者(Producer):创建并发送消息的JMS客户端,在点对点模型中就是发送者,在发布/订阅模型中就是发布者。
• 消费者(Consumer):接收消息的JMS客户端,在点对点模型中就是接收者,在发布/订阅模型中就是订阅者。
• 客户端(Client):生产或消费消息的基于Java的应用程序或对象。
• 队列(Queue ):一个容纳被发送的等待阅读的消息的区域。它是点对点模型中的队列。
• 主题(Topic):一种支持发送消息给多个订阅者的机制。它是发布/订阅模型中的主题。
• 消息(Message):在JMS客户端之间传递的数据对象。JMS消息又包括消息头、属性和消息体三部分。

第3章RabbitMQ

第4章ActiveMQ

第5章Kafka

第6章RocketMC