SpringBoot+MQTT+apollo实现订阅发布功能

旧城等待， 2022-09-05 00:16 206阅读 0赞

**点击关注公众号，利用****碎片时间学习****![3e4cd68b2e511bb0c955a1fa113a27db.png][]**

由于最近公司在开发一款后台与安卓的更新系统，经过再三研究之后，也是选择Mqtt这个目前流行的框架。为了能够让项目运营起来，最终虽说是选择ActiveMQ。

但在这个过程中，也是发现Apollo作为服务器也是相当不错。当然对于后者已经被apace放弃，不过今天还是和大家整理一下SpringBoot+MQTT+apollo实现订阅发布功能的全过程。

对于项目首先需要用到的前提东西，比如Apollo如何下载，以及MQTT测试工具在这里就不多说。如果真的不懂私聊Damon吧，在这里就不浪费时间。

对于项目，首先你所需要引入maven包：

#### pom.xml ####

<dependency>
        <groupId>org.springframework.integration</groupId>
        <artifactId>spring-integration-stream</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.integration</groupId>
        <artifactId>spring-integration-mqtt</artifactId>
    </dependency>

其目标就是将MQTT用在项目组中

接着就是项目yml文件的配置，使用properties，以葫芦画瓢就行了：

#### applicaiton.yml ####

mqtt:
      username: admin
      password: password
      host-url: tcp://127.0.0.1:8161  # 你自己服务器的地址和端口，这个需要改
      clientID: test1       # 这个改不改随意，但不同的客户端肯定不能一样
      default-topic: home/garden/fountain      # 默认主题
      timeout: 100
      keepalive: 100
    
    # Tomcat
    server:
      tomcat:
        uri-encoding: UTF-8
        max-threads: 1000
        min-spare-threads: 30
      port: 8088

注意host-url，这就是你apollo的地址

来到第三步，此时就是项目内的文件：

#### MqttConfig文件 ####

@Component
    @ConfigurationProperties("mqtt")
    @Setter
    @Getter
    public class MqttConfig {
        @Autowired
        private MqttPushClient mqttPushClient;
    
        /**
         * 用户名
         */
        // @Value("username")
        private String username;
        /**
         * 密码
         */
        private String password;
        /**
         * 连接地址
         */
        private String hostUrl;
        /**
         * 客户Id
         */
        private String clientID;
        /**
         * 默认连接话题
         */
        private String defaultTopic;
        /**
         * 超时时间
         */
        private int timeout;
        /**
         * 保持连接数
         */
        private int keepalive;
    
        @Bean
        public MqttPushClient getMqttPushClient() {
            System.out.println("hostUrl: "+ hostUrl);
            System.out.println("clientID: "+ clientID);
            System.out.println("username: "+ username);
            System.out.println("password: "+ password);
            System.out.println("timeout: "+timeout);
            System.out.println("keepalive: "+ keepalive);
            mqttPushClient.connect(hostUrl, clientID, username, password, timeout, keepalive);
            // 以/#结尾表示订阅所有以test开头的主题
            mqttPushClient.subscribe(defaultTopic, 0);
            return mqttPushClient;
        }
    }

目的就是配置所对应的消息

第四步就是发布以及订阅等功能：

#### MqttPushClient ####

@Component
    public class MqttPushClient {
        private static final Logger logger = LoggerFactory.getLogger(MqttPushClient.class);
    
        @Autowired
        private PushCallback pushCallback;
    
        private static MqttClient client;
    
        private static MqttClient getClient() {
            return client;
        }
    
        private static void setClient(MqttClient client) {
            MqttPushClient.client = client;
        }
    
        /**
         * 客户端连接
         *
         * @param host      ip+端口
         * @param clientID  客户端Id
         * @param username  用户名
         * @param password  密码
         * @param timeout   超时时间
         * @param keepalive 保留数
         */
        public void connect(String host, String clientID, String username, String password, int timeout, int keepalive) {
            MqttClient client;
            try {
                client = new MqttClient(host, clientID, new MemoryPersistence());
                MqttConnectOptions options = new MqttConnectOptions();
                options.setCleanSession(true);
                options.setUserName(username);
                options.setPassword(password.toCharArray());
                options.setConnectionTimeout(timeout);
                options.setKeepAliveInterval(keepalive);
                MqttPushClient.setClient(client);
                try {
                    client.setCallback(pushCallback);
                    client.connect(options);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    
        /**
         * 发布
         *
         * @param qos         连接方式
         * @param retained    是否保留
         * @param topic       主题
         * @param pushMessage 消息体
         */
        public void publish(int qos, boolean retained, String topic, String pushMessage) {
            MqttMessage message = new MqttMessage();
            message.setQos(qos);
            message.setRetained(retained);
            message.setPayload(pushMessage.getBytes());
            MqttTopic mTopic = MqttPushClient.getClient().getTopic(topic);
            if (null == mTopic) {
                logger.error("topic not exist");
            }
            MqttDeliveryToken token;
            try {
                token = mTopic.publish(message);
                token.waitForCompletion();
            } catch (MqttPersistenceException e) {
                e.printStackTrace();
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }
    
        /**
         * 订阅某个主题
         *
         * @param topic 主题
         * @param qos   连接方式
         */
        public void subscribe(String topic, int qos) {
            logger.info("开始订阅主题" + topic);
            try {
                MqttPushClient.getClient().subscribe(topic, qos);
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }
    }

订阅主题以及发布的方式等内容更多编写

最后在搞个测试看看我们的结果是否正确：

#### TestController ####

@RestController
    @RequestMapping("/")
    public class TestController {
    
        @Autowired
        private MqttPushClient mqttPushClient;
    
        @GetMapping(value = "/publishTopic")
        public String publishTopic() {
            String topicString = "home/garden/fountain";
            mqttPushClient.publish(0, false, topicString, "测试一下发布消息");
            return "ok";
        }
        // 发送自定义消息内容（使用默认主题）
        @RequestMapping("/publishTopic/{data}")
        public String test1(@PathVariable("data") String data) {
            String topicString = "home/garden/fountain";
            mqttPushClient.publish(0,false,topicString, data);
            return "ok";
        }
    
        // 发送自定义消息内容，且指定主题
        @RequestMapping("/publishTopic/{topic}/{data}")
        public String test2(@PathVariable("topic") String topic, @PathVariable("data") String data) {
            mqttPushClient.publish(0,false,topic, data);
            return "ok";
        }
    }

如此一来就OK！

你可以使用MQTT.fx进行测试。用Postman发出，就能够查看最终的结果。在这里，因为时间的原因就不多说，有啥有趣的问题，咱们可以一同探讨。

希望你希望，Damon将会不断的分享各种有趣的开发小故事给大家娱乐。下一期，或在POI实现导出导入或者是ActiveMQ进行选择。

*（感谢阅读，希望对你所有帮助)*

*来源：blog.csdn.net/qq\_41689567/article/details/106774202*

推荐好文

[强大，10k+点赞的 SpringBoot 后台管理系统竟然出了详细教程！][10k_ SpringBoot]

[分享一套基于SpringBoot和Vue的企业级中后台开源项目，代码很规范！][SpringBoot_Vue]

[能挣钱的，开源 SpringBoot 商城系统，功能超全，超漂亮][SpringBoot]![58a127df96dd49150bedb22530780a51.png][]

[3e4cd68b2e511bb0c955a1fa113a27db.png]: /images/20220829/c205a06e243945c5b8013b9b0a2069d9.png
[10k_ SpringBoot]: http://mp.weixin.qq.com/s?__biz=MzI4Njc5NjM1NQ%3D%3D&chksm=ebd5c411dca24d07446a4f22772cd12d2cc1143b62eb922a14500ae981cd57fb850ef74ca257&idx=2&mid=2247498557&scene=21&sn=84902d7030e174e152615d8fae3755cc#wechat_redirect
[SpringBoot_Vue]: http://mp.weixin.qq.com/s?__biz=MzU2MTI4MjI0MQ%3D%3D&chksm=fc799be6cb0e12f0b58479f8b031bc5ae79efdca1880ff9b19f0a107c3a88c37b9596dc73203&idx=2&mid=2247497288&scene=21&sn=65d7111a19e920df1e9416959594b2f4#wechat_redirect
[SpringBoot]: http://mp.weixin.qq.com/s?__biz=MzU2MTI4MjI0MQ%3D%3D&chksm=fc799a9ccb0e138a64a2671f31b8d4be1b3c061a288f7b0d1d4af02a60c45a72231446b91e00&idx=2&mid=2247497522&scene=21&sn=c5b4e596be4ac7d7f293e7e2f0ea0769#wechat_redirect
[58a127df96dd49150bedb22530780a51.png]: /images/20220829/3247f3e1272f44b9bcb460416e442b34.png