MQTT - broker - 本地安装启动moquette

本是古典何须时尚 2022-12-25 01:51 175阅读 0赞

# 一、MQTT简介 #

MQTT（Message Queuing Telemetry Transport，消息队列遥测传输）是一个基于客户端-服务端的消息发布/订阅传输协议。它是一种基于发布/订阅（publish/subscribe）模式的"轻量级"构建于TCP/IP协议上的通讯协议，该协议构建于TCP/IP协议上。MQTT最大优点在于，可以以极少的代码和有限的带宽，为连接远程设备提供实时可靠的消息服务。作为一种低开销、低带宽占用的即时通讯协议，其在物联网、小型设备、移动应用等方面有着较广泛的应用。

# 二、MQTT Broker #

broker的主要职责是接受发布者发布的所有消息，并将其过滤后分发给不同的消息订阅者。

MQTT支持众多broker实现方式，具体参见MQTT官网说明：[https://mqtt.org/software/][https_mqtt.org_software]，broker对比参见[https://www.jianshu.com/p/cf91f4bea071][https_www.jianshu.com_p_cf91f4bea071],为方便阅读，从文档摘图如下：(若有侵权请作者联系我删除)：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N4ZzAyMDU_size_16_color_FFFFFF_t_70][]

本文主要介绍 moquette 的本地安装启动方式；

# 三、Moquette #

## 3.1 源码地址 ##

本文样例代码地址；[https://github.com/andsel/moquette][https_github.com_andsel_moquette]

其他git的地址参见: [https://github.com/milliondreams/moquette-mqtt][https_github.com_milliondreams_moquette-mqtt]

## 3.2 代码架构 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N4ZzAyMDU_size_16_color_FFFFFF_t_70 1][]

## 3.3 本地启动broker ##

将项目代码导入开发工具(本文使用idea工具)，服务启动类代码参见：io.moquette.broker.Server

代码如下：

/*
     * Copyright (c) 2012-2018 The original author or authors
     * ------------------------------------------------------
     * All rights reserved. This program and the accompanying materials
     * are made available under the terms of the Eclipse Public License v1.0
     * and Apache License v2.0 which accompanies this distribution.
     *
     * The Eclipse Public License is available at
     * http://www.eclipse.org/legal/epl-v10.html
     *
     * The Apache License v2.0 is available at
     * http://www.opensource.org/licenses/apache2.0.php
     *
     * You may elect to redistribute this code under either of these licenses.
     */
    package io.moquette.broker;
    
    import io.moquette.BrokerConstants;
    import io.moquette.broker.config.*;
    import io.moquette.interception.InterceptHandler;
    import io.moquette.persistence.H2Builder;
    import io.moquette.persistence.MemorySubscriptionsRepository;
    import io.moquette.interception.BrokerInterceptor;
    import io.moquette.broker.security.*;
    import io.moquette.broker.subscriptions.CTrieSubscriptionDirectory;
    import io.moquette.broker.subscriptions.ISubscriptionsDirectory;
    import io.moquette.broker.security.IAuthenticator;
    import io.moquette.broker.security.IAuthorizatorPolicy;
    import io.netty.handler.codec.mqtt.MqttPublishMessage;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import java.io.File;
    import java.io.IOException;
    import java.lang.reflect.InvocationTargetException;
    import java.text.ParseException;
    import java.util.*;
    import java.util.concurrent.Executors;
    import java.util.concurrent.ScheduledExecutorService;
    
    import static io.moquette.logging.LoggingUtils.getInterceptorIds;
    
    public class Server {
    
        private static final Logger LOG = LoggerFactory.getLogger(io.moquette.broker.Server.class);
    
        private ScheduledExecutorService scheduler;
        private NewNettyAcceptor acceptor;
        private volatile boolean initialized;
        private PostOffice dispatcher;
        private BrokerInterceptor interceptor;
        private H2Builder h2Builder;
        private SessionRegistry sessions;
    
        public static void main(String[] args) throws IOException {
            final Server server = new Server();
            server.startServer();
            System.out.println("Server started, version 0.13-SNAPSHOT");
            //Bind a shutdown hook
            Runtime.getRuntime().addShutdownHook(new Thread(server::stopServer));
        }
    
        /**
         * Starts Moquette bringing the configuration from the file located at m_config/moquette.conf
         *
         * @throws IOException in case of any IO error.
         */
        public void startServer() throws IOException {
            File defaultConfigurationFile = defaultConfigFile();
            LOG.info("Starting Moquette integration. Configuration file path={}", defaultConfigurationFile.getAbsolutePath());
            IResourceLoader filesystemLoader = new FileResourceLoader(defaultConfigurationFile);
            final IConfig config = new ResourceLoaderConfig(filesystemLoader);
            startServer(config);
        }
    
        private static File defaultConfigFile() {
            String configPath = System.getProperty("moquette.path", null);
            return new File(configPath, IConfig.DEFAULT_CONFIG);
        }
    
        /**
         * Starts Moquette bringing the configuration from the given file
         *
         * @param configFile text file that contains the configuration.
         * @throws IOException in case of any IO Error.
         */
        public void startServer(File configFile) throws IOException {
            LOG.info("Starting Moquette integration. Configuration file path: {}", configFile.getAbsolutePath());
            IResourceLoader filesystemLoader = new FileResourceLoader(configFile);
            final IConfig config = new ResourceLoaderConfig(filesystemLoader);
            startServer(config);
        }
    
        /**
         * Starts the integration with the given properties.
         * <p>
         * Its suggested to at least have the following properties:
         * <ul>
         *  <li>port</li>
         *  <li>password_file</li>
         * </ul>
         *
         * @param configProps the properties map to use as configuration.
         * @throws IOException in case of any IO Error.
         */
        public void startServer(Properties configProps) throws IOException {
            LOG.debug("Starting Moquette integration using properties object");
            final IConfig config = new MemoryConfig(configProps);
            startServer(config);
        }
    
        /**
         * Starts Moquette bringing the configuration files from the given Config implementation.
         *
         * @param config the configuration to use to start the broker.
         * @throws IOException in case of any IO Error.
         */
        public void startServer(IConfig config) throws IOException {
            LOG.debug("Starting Moquette integration using IConfig instance");
            startServer(config, null);
        }
    
        /**
         * Starts Moquette with config provided by an implementation of IConfig class and with the set
         * of InterceptHandler.
         *
         * @param config   the configuration to use to start the broker.
         * @param handlers the handlers to install in the broker.
         * @throws IOException in case of any IO Error.
         */
        public void startServer(IConfig config, List<? extends InterceptHandler> handlers) throws IOException {
            LOG.debug("Starting moquette integration using IConfig instance and intercept handlers");
            startServer(config, handlers, null, null, null);
        }
    	
    	... ...
    }

为方便阅读，此处仅摘取部分代码显示，若有其他需要，请自行下载全部代码查阅；

阅读代码可知，启动服务需依赖读取moquette配置文件：

![20201130104457765.png][]

默认读取的为 IConfig.DEFAULT\_CONFIG 配置的 "config/moquette.conf"，即 moquette.path 配置指向代码 config 上级目录即可；

moquette.path 本地配置方式如下：

![2020113010323784.png][]

![2020113010331830.png][]

配置完成后右键 RUN Server.main()，查看启动日志：

![20201130103729946.png][]

至此 moquette 本地服务启动成功；

# 四 订阅发布demo #

基于 MQTT-Client实现，参见：[MQTT - 订阅发布demo][MQTT - _demo]

[https_mqtt.org_software]: https://mqtt.org/software/
[https_www.jianshu.com_p_cf91f4bea071]: https://www.jianshu.com/p/cf91f4bea071
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N4ZzAyMDU_size_16_color_FFFFFF_t_70]: /images/20221120/62e77f5ce31e467ead944b0de5aade53.png
[https_github.com_andsel_moquette]: https://github.com/andsel/moquette
[https_github.com_milliondreams_moquette-mqtt]: https://github.com/milliondreams/moquette-mqtt
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N4ZzAyMDU_size_16_color_FFFFFF_t_70 1]: /images/20221120/1a8bd6d36f9d4517adde384f263a3e78.png
[20201130104457765.png]: /images/20221120/e1ff2daeb43b4e518c62ea62e8984054.png
[2020113010323784.png]: /images/20221120/d2d5d0a804d545b789d8027ef218fee9.png
[2020113010331830.png]: /images/20221120/b7a16aaa0c0546dbb18e72b931424e81.png
[20201130103729946.png]: /images/20221120/6a230f25b456435a96f6f6a55ad07444.png
[MQTT - _demo]: https://blog.csdn.net/sxg0205/article/details/110382966