物联网：七天构建一个闭环的物联网DEMO-MQTT的配置

这篇文章我们看看JAVA 中 MQTT 的连接配置， 先看代码：

/*
 * Copyright 2022 Pnoker All Rights Reserved
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.ruoyi.project.business.controlcenter.iot.config;

import com.ruoyi.project.business.controlcenter.iot.util.IpUtil;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

import javax.annotation.Resource;
import java.security.InvalidKeyException;
import java.security.NoSuchAlgorithmException;


/**
 * @author pnoker
 */
@Slf4j
@Configuration
@IntegrationComponentScan(basePackages = "com.ruoyi.project.business.controlcenter.iot")
@EnableConfigurationProperties({MqttProperties.class})
public class MqttConfig {

    private static final String RANDOM_ID = IpUtil.getOneLocalIP(); //CommonConstant.SYMBOL.UNDERSCORE + RandomUtil.randomString(8);

    @Resource
    private MqttProperties mqttProperties;

    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageChannel mqttOutputChannel() {
        return new DirectChannel();
    }

    /**
     * MqttPahoMessageDrivenChannelAdapter 本质上是 MessageProducer
     */
    @Bean
    public MqttPahoMessageDrivenChannelAdapter messageProducer(MessageChannel mqttInputChannel) throws NoSuchAlgorithmException, InvalidKeyException {
        String clientId = mqttProperties.getClient() + "_sub" + RANDOM_ID;
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(
                        clientId, mqttClientFactory("_sub"));
        log.info("mqtt-client-sub inited , clientId : {}...", clientId);
        adapter.setQos(mqttProperties.getQos());
        adapter.setOutputChannel(mqttInputChannel);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setCompletionTimeout(mqttProperties.getCompletionTimeout());
        return adapter;
    }

//    @Bean
//    @ServiceActivator(inputChannel = "mqttInputChannel")
//    public MessageHandler inputbound() {
//        return new MqttReceiveHander();
//        // TODO 先不做收消息
//        return null;
//    }

    @Bean
    @ServiceActivator(inputChannel = "mqttOutputChannel")
    public MessageHandler outbound() throws NoSuchAlgorithmException, InvalidKeyException {
        String client = mqttProperties.getClient() + "_pub" + RANDOM_ID;
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(client, mqttClientFactory("_pub"));
        messageHandler.setAsync(true);
        log.info("mqtt-client-pub inited , clientId : {}...", client);
        return messageHandler;
    }

    public MqttPahoClientFactory mqttClientFactory(String suffix) throws NoSuchAlgorithmException, InvalidKeyException {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setConnectionOptions(getMqttConnectOptions(suffix));
        return factory;
    }

    public MqttConnectOptions getMqttConnectOptions(String suffix) throws NoSuchAlgorithmException, InvalidKeyException {
        MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();

        // username & password
        if (mqttProperties.getAuthType().equals(MqttProperties.AuthTypeEnum.USERNAME)) {
            mqttConnectOptions.setUserName(mqttProperties.getUsername());
            mqttConnectOptions.setPassword(mqttProperties.getPassword().toCharArray());
        } else if (mqttProperties.getAuthType().equals(MqttProperties.AuthTypeEnum.X509)) {
            // tls x509
            throw new IllegalArgumentException("暂不支持的认证方式");
        } else if (mqttProperties.getAuthType().equals(MqttProperties.AuthTypeEnum.ALI)) {
            ConnectionOptionWrapper connectionOptionWrapper = new ConnectionOptionWrapper(mqttProperties.getInstanceId(), mqttProperties.getAccessKey(), mqttProperties.getSecretKey(), mqttProperties.getClient() + suffix + RANDOM_ID);
            mqttConnectOptions = connectionOptionWrapper.getMqttConnectOptions();
        }

        // disable https hostname verification
        mqttConnectOptions.setHttpsHostnameVerificationEnabled(false);
        mqttConnectOptions.setServerURIs(new String[]{mqttProperties.getUrl()});
        mqttConnectOptions.setKeepAliveInterval(mqttProperties.getKeepAlive());
        return mqttConnectOptions;

    }

}

这里比较重要的一个地方是 getMqttConnectOptions ， 它里面主要是根据你的 MQTT 的安全策略定制相应的连接策略， 运行的时候可以稍微注意下， 如果遇到连接不上报错的地方 ， 就要检查下MQTT 的安装策略是否匹配。

（

开源地址：

wowiot: 基于ruoyi二开的一款开源智能硬件管理平台， 当前可支持硬件管理，并且支持 esp32 定制的api，可实现远程开关信号的下发，更多功能欢迎探索，有更多的功能性需求，欢迎入群讨论或者参与进来

）