mosquitto-1.4.5的安装和使用

小咪咪 2022-05-15 16:50 215阅读 0赞

## 安装 ##

*  下载程序文件

pwd
    /usr/local
    wget http://mosquitto.org/files/source/mosquitto-1.4.5.tar.gz

*  解压程序文件

tar -xzvf mosquitto-1.4.5.tar.gz 
    cd mosquitto-1.4.5.tar.gz

*  关闭不需要的安装选项

> config.mk包括了多个选项, 可按需关闭或开启，但一旦开启则需要先安装对应的模块

vi config.mk
    
    # Build with SRV lookup support.
    WITH_SRV:=no
    
    # Build using libuuid for clientid generation (Linux only - please report if
    # supported on your platform).
    WITH_UUID:=no
    
    # Build with websockets support on the broker.
    WITH_WEBSOCKETS:=no

*  安装基础软件

yum install openssl-devel
    yum install gcc-c++
    yum install cmake
    yum install libuuid-devel  （WITH_UUID:=yes 需要）

*  安装程序

make 
    make install

## 启动与停止 ##

*  创建用户

> mosquitto默认以mosquitto用户启动，可以通过配置文件修改

groupadd mosquitto
    useradd -g mosquitto mosquitto

*  设置用户和密码

mosquitto_passwd -c pwfile.example tianming（此为用户名，然后输入两次密码）

*  程序配置

mv /etc/mosquitto/mosquitto.conf.example /etc/mosquitto/mosquitto.conf

*  启动

> 成功将启动1883端口监听  
> mosquitto -c /etc/mosquitto/mosquitto.conf -d

[root@localhost mosquitto]# pwd
    /etc/mosquitto
    [root@localhost mosquitto]# mosquitto -c /etc/mosquitto/mosquitto.conf -d
    [root@localhost mosquitto]# 1541836831: mosquitto version 1.4.5 (build date 2018-11-09 23:43:35-0800) starting
    1541836831: Config loaded from /etc/mosquitto/mosquitto.conf.
    1541836831: Opening ipv4 listen socket on port 1883.
    1541836831: Opening ipv6 listen socket on port 1883.

*  停止

[root@localhost ~]# ps -ef|grep mosquitto
    mosquit+  23929      1  0 23:52 pts/2    00:00:00 mosquitto -c /etc/mosquitto/mosquitto.conf -d
    root      24427  24084  0 23:59 pts/3    00:00:00 grep --color=auto mosquitto
    [root@localhost ~]# kill -9 23929
    [root@localhost ~]#

## CentOS 6.5关闭防火墙 ##

[root@localhost ~]#servcie iptables stop --临时关闭防火墙
    
    [root@localhost ~]#chkconfig iptables off --永久关闭防火墙

## 使用 ##

*  [Mosquitto的简单使用][Mosquitto]
 *  启动MQTT

mosquitto -c /etc/mosquitto/mosquitto.conf -d

## 介绍 ##

*  [Mosquitto 介绍][Mosquitto 1]

## 配置 ##

*  application-dev.properties

#MQTT
    #MQTT-用户名
    spring.mqtt.username=tianming
    #MQTT-密码
    spring.mqtt.password=tianming
    #MQTT-服务器连接地址，如果有多个，用逗号隔开
    spring.mqtt.url=tcp://192.168.122.29:1883
    #MQTT-连接服务器默认客户端ID
    spring.mqtt.client.id=65515
    #MQTT-默认的消息推送主题，实际可在调用接口时指定
    spring.mqtt.default.topic=topic
    #超时时间，单位为秒
    spring.mqtt.connectionTimeout=30
    #会话心跳时间，单位为秒
    spring.mqtt.keepAliveInterval=30
    #MQTT主题
    spring.mqtt.theme=MS/KD/65515
    
    #AES128秘钥
    spring.aes.key=asdfgjqwerio
    
    # destination file directory
    spring.destination.orderpush=/app/rookie

## 源码 ##

> SpringBoot2.0集成MQTT实现消息推送功能

*  配置类MqttSenderConfig

import org.springframework.beans.factory.annotation.Value;
    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.outbound.MqttPahoMessageHandler;
    import org.springframework.messaging.MessageChannel;
    import org.springframework.messaging.MessageHandler;
    
    @Configuration
    @IntegrationComponentScan
    public class MqttSenderConfig { 
    	@Value("${spring.mqtt.username}")
    	private String username;
    	@Value("${spring.mqtt.password}")
    	private String password;
    	@Value("${spring.mqtt.url}")
    	private String hostUrl;
    	@Value("${spring.mqtt.client.id}")
    	private String clientId;
    	@Value("${spring.mqtt.default.topic}")
    	private String defaultTopic;
    	@Value("${spring.mqtt.connectionTimeout}")
    	private int connectionTimeout;
    	@Value("${spring.mqtt.keepAliveInterval}")
    	private int keepAliveInterval;
    
    	@Bean
    	public MqttPahoClientFactory mqttClientFactory() { 
    		DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
    		factory.setUserName(username);
    		factory.setPassword(password);
    		factory.setServerURIs(new String[] {  hostUrl });
    		// 设置超时时间
    		factory.setConnectionTimeout(connectionTimeout);
    		// 设置会话心跳时间
    		factory.setKeepAliveInterval(keepAliveInterval);
    		return factory;
    	}
    
    	@Bean
    	@ServiceActivator(inputChannel = "mqttOutboundChannel")
    	public MessageHandler mqttOutbound() { 
    		MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(clientId, mqttClientFactory());
    		messageHandler.setAsync(true);
    		messageHandler.setDefaultTopic(defaultTopic);
    		return messageHandler;
    	}
    
    	@Bean
    	public MessageChannel mqttOutboundChannel() { 
    		return new DirectChannel();
    	}
    }

*  接口类MqttGateway

import org.springframework.integration.annotation.MessagingGateway;
    import org.springframework.integration.mqtt.support.MqttHeaders;
    import org.springframework.messaging.handler.annotation.Header;
    import org.springframework.stereotype.Component;
    
    @Component
    @MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
    public interface MqttGateway { 
    	void sendToMqtt(String data);
    	void sendToMqtt(String data, @Header(MqttHeaders.TOPIC) String topic);
    	// data:发送内容 topic:主题名称 qos:消息处理机制（值为：0,1,2，默认为0）
    	void sendToMqtt(String data, @Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos);
    }

*  目标类调用

int num = 0;
    for (int i = 0; i < files.length; i++) { 
    	InputStreamReader inReader = new InputStreamReader(new FileInputStream(files[i]), "GBK");
    	BufferedReader bReader = new BufferedReader(inReader);
    	contentLine = bReader.readLine();
    	while (contentLine != null) { 
    		// 日志打印
    		LOGGER.info("读取的当前数据==》" + contentLine.trim());
    		// AES加密
    		String encryptContent = AESCoder.encrypt(aesKey, contentLine.trim());
    		// 发送到MQTT
    		mqttGateway.sendToMqtt(encryptContent, mqttTheme);
    		num++;
    		// 读取下一行
    		contentLine = bReader.readLine();
    	}
    	bReader.close();
    }

## MQTT主题 ##

*  topic是消息发送的主题,这里可以自己灵活定义,也可以使用默认的主题,就是配置文件的主题,mqtt\_qos是mqtt 对消息处理的几种机制分为0,1,2 其中0表示的是订阅者没收到消息不会再次发送,消息会丢失,1表示的是会尝试重试,一直到接收到消息,但这种情况可能导致订阅者收到多次重复消息,2相比多了一次去重的动作,确保订阅者收到的消息有一次
 *  当然,这三种模式下的性能肯定也不一样,mqtt\_qos=0是最好的,2是最差的

## 集成 ##

*  [SpringBoot2.0集成MQTT消息推送功能][SpringBoot2.0_MQTT]
 *  [SpringBoot2.0如何集成mqtt消息推送][SpringBoot2.0_mqtt]
 *  [SpringBoot2.0集成MQTT功能之消息订阅处理][SpringBoot2.0_MQTT 1]

--------------------

参考博客：  
\[1\]. [https://www.cnblogs.com/littleatp/p/4835879.html][https_www.cnblogs.com_littleatp_p_4835879.html]

[Mosquitto]: http://www.cnblogs.com/yueli/p/7478543.html
[Mosquitto 1]: http://www.cnblogs.com/yueli/p/7478449.html
[SpringBoot2.0_MQTT]: https://blog.csdn.net/qq_41018959/article/details/80592444
[SpringBoot2.0_mqtt]: https://blog.csdn.net/zhangxing52077/article/details/80568244
[SpringBoot2.0_MQTT 1]: https://blog.csdn.net/qq_41018959/article/details/81076055
[https_www.cnblogs.com_littleatp_p_4835879.html]: https://www.cnblogs.com/littleatp/p/4835879.html