public void 前端業(yè)務(wù)/硬件業(yè)務(wù)()
{
    刷新界面();
    更新數(shù)據(jù)庫(kù)();
    對(duì)界面更新數(shù)據(jù)();
    ………………………………
}

.\apollo.cmd create mybroker2

.\apollo-broker.cmd run

<dependencies>
    <!--導(dǎo)入起步依賴-->
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-web</artifactId>
		</dependency>
			<artifactId>spring-boot-starter-integration</artifactId>
			<groupId>org.springframework.integration</groupId>
			<artifactId>spring-integration-stream</artifactId>
			<artifactId>spring-integration-mqtt</artifactId>
			<groupId>org.mybatis.spring.boot</groupId>
			<artifactId>mybatis-spring-boot-starter</artifactId>
			<version>1.3.0</version>
			<artifactId>spring-boot-configuration-processor</artifactId>
			<optional>true</optional>
			<groupId>com.alibaba</groupId>
			<artifactId>fastjson</artifactId>
			<version>1.2.47</version>
		
</dependencies>

#mqtt配置
com:
  mqtt:
    url: tcp://127.0.0.1:61613
    clientId: mqtt_test1234
    topics: topic01,topic02
    username: admin
    password: password
    timeout: 10
    keepalive: 20
    
#指定服務(wù)端口 
server:
  port: 8081   #一般沒(méi)改過(guò)tomcat服務(wù)器的端口不用修改

package com.vcarecity.config;
 
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;
/**
 * 讀取yml
 */
@Component
@ConfigurationProperties(prefix = "com.mqtt")     //對(duì)應(yīng)yml文件中的com下的mqtt文件配置
public class MqttConfiguration {
	private String url;
	private String clientId;
	private String topics;
    private String username;
    private String password;
    private String timeout;
    private String keepalive;
	public String getUrl() {
		return url;
	}
	public void setUrl(String url) {
		this.url = url;
	public String getUsername() {
		return username;
	public void setUsername(String username) {
		this.username = username;
	public String getPassword() {
		return password;
	public void setPassword(String password) {
		this.password = password;
	public String getClientId() {
		return clientId;
	public void setClientId(String clientId) {
		this.clientId = clientId;
	public String getTopics() {
		return topics;
	public void setTopics(String topics) {
		this.topics = topics;
	public String getTimeout() {
		return timeout;
	public void setTimeout(String timeout) {
		this.timeout = timeout;
	public String getKeepalive() {
		return keepalive;
	public void setKeepalive(String keepalive) {
		this.keepalive = keepalive;
}

package com.vcarecity.mqtt;
 
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
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;
import com.vcarecity.config.MqttConfiguration;
/**
 * MQTT生產(chǎn)端
 *
 */
@Configuration
public class MqttOutboundConfiguration {
	@Autowired
	private MqttConfiguration mqttProperties;
	@Bean
	public MessageChannel mqttOutboundChannel() {
		return new DirectChannel();
	}
	public MqttPahoClientFactory mqttClientFactory() {
		DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
		String[] array = mqttProperties.getUrl().split(",");
		MqttConnectOptions options = new MqttConnectOptions();
		options.setServerURIs(array);
		options.setUserName(mqttProperties.getUsername());
		options.setPassword(mqttProperties.getPassword().toCharArray());
		// 接受離線消息
		options.setCleanSession(false); //告訴代理客戶端是否要建立持久會(huì)話   false為建立持久會(huì)話
		factory.setConnectionOptions(options);
		return factory;
	@ServiceActivator(inputChannel = "mqttOutboundChannel")
	public MessageHandler mqttOutbound() {
		MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(
				mqttProperties.getClientId()+"outbound", mqttClientFactory());
		messageHandler.setAsync(true);
		return messageHandler;
}

package com.vcarecity.mqtt;
 
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Autowired;
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.core.MessageProducer;
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.support.DefaultPahoMessageConverter;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import com.vcarecity.config.MqttConfiguration;
/**
 * MQTT消費(fèi)端
 *
 */
@Configuration
@IntegrationComponentScan
public class MqttInboundConfiguration {
	@Autowired
	private MqttConfiguration mqttProperties;
	@Bean
	public MessageChannel mqttInputChannel() {
		return new DirectChannel();
	}
	public MqttPahoClientFactory mqttClientFactory() {
		DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
		    String[] array = mqttProperties.getUrl().split(",");
			MqttConnectOptions options = new MqttConnectOptions();
			options.setServerURIs(array);
			options.setUserName(mqttProperties.getUsername());
			options.setPassword(mqttProperties.getPassword().toCharArray());
			options.setKeepAliveInterval(2);
			//接受離線消息
			options.setCleanSession(false);
			factory.setConnectionOptions(options);
		return factory;
	//配置client,監(jiān)聽(tīng)的topic
	public MessageProducer inbound() {
		String[] inboundTopics = mqttProperties.getTopics().split(",");
		MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(
				mqttProperties.getClientId()+"_inbound",mqttClientFactory(), inboundTopics);  //對(duì)inboundTopics主題進(jìn)行監(jiān)聽(tīng)
		adapter.setCompletionTimeout(5000);
		adapter.setQos(1);
		adapter.setConverter(new DefaultPahoMessageConverter());
		adapter.setOutputChannel(mqttInputChannel());
		return adapter;
	//通過(guò)通道獲取數(shù)據(jù)
	@ServiceActivator(inputChannel = "mqttInputChannel")  //異步處理
	public MessageHandler handler() {
		return new MessageHandler() {
			@Override
			public void handleMessage(Message<?> message) throws MessagingException {
//				System.out.println("message:"+message);
				System.out.println("----------------------");
				System.out.println("message:"+message.getPayload());
				System.out.println("PacketId:"+message.getHeaders().getId());
				System.out.println("Qos:"+message.getHeaders().get(MqttHeaders.QOS));
				String topic = (String) message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC);
				System.out.println("topic:"+topic);
			}
		};
}

package com.vcarecity.controller;
 
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import com.vcarecity.mqtt.MqttGateway;
@RestController
public class MqttPubController {
	
	@SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
	@Autowired
	private MqttGateway mqttGateway;
	@RequestMapping("/hello")
	public String hello() {
		return "hello!";
	}
	 @RequestMapping("/sendMqtt")
	    public String sendMqtt(String  sendData){
		 	System.out.println(sendData);
	    	System.out.println("進(jìn)入sendMqtt-------"+sendData);
	        mqttGateway.sendToMqtt("topic01",(String) sendData);
	        return "Test is OK";
	    }
	@RequestMapping("/sendMqttTopic")
	public String sendMqtt(String  sendData,String topic){
		//System.out.println(sendData+"   "+topic);
		//System.out.println("進(jìn)入inbound發(fā)送："+sendData);
		mqttGateway.sendToMqtt(topic,(String) sendData);
		return "Test is OK";
}

//通道2
@Bean
public MessageChannel mqttInputChannelTwo() {
    return new DirectChannel();
}
//配置client2，監(jiān)聽(tīng)的topic:hell2,hello3
public MessageProducer inbound1() {
    MqttPahoMessageDrivenChannelAdapter adapter =
            new MqttPahoMessageDrivenChannelAdapter(clientId+"_inboundTwo", mqttClientFactory(),
                    "hello2","hello3");
    adapter.setCompletionTimeout(completionTimeout);
    adapter.setConverter(new DefaultPahoMessageConverter());
    adapter.setQos(1);
    adapter.setOutputChannel(mqttInputChannelTwo());
    return adapter;
 
//通過(guò)通道2獲取數(shù)據(jù)
@ServiceActivator(inputChannel = "mqttInputChannelTwo")
public MessageHandler handlerTwo() {
    return new MessageHandler() {
       @Override
			public void handleMessage(Message<?> message) throws MessagingException {
//				System.out.println("message:"+message);
				System.out.println("----------------------");
				System.out.println("message:"+message.getPayload());
				System.out.println("PacketId:"+message.getHeaders().getId());
				System.out.println("Qos:"+message.getHeaders().get(MqttHeaders.QOS));
				String topic = (String) message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC);
				System.out.println("topic:"+topic);
    };

topics: topic03,topic04,topic01,topic02