mqtt消息推送-推送服务简单实现

上一章已经部署了mosquito服务器。这里实现一个简单的推送服务功能。主要包括两个部分。一个上传推送消息的API接口。连接mosquito并推送消息的服务。

简单的例子可以使用springboot快速开发，使用默认配置即可。
新建maven项目mqtt-server，pom.xml配置文件如下：

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <groupId>com.mob</groupId>
  <artifactId>mqtt-server</artifactId>
  <packaging>jar</packaging>
  <version>0.0.1-SNAPSHOT</version>
  <name>mqtt-server Maven Webapp</name>
  <url>http://maven.apache.org</url>

  <parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>1.5.3.RELEASE</version>
    <relativePath/>
  </parent>

  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <java.version>1.8</java.version>
  </properties>

  <dependencies>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
      <groupId>org.eclipse.paho</groupId>
      <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
      <version>1.1.0</version>
    </dependency>
    <dependency>
      <groupId>junit</groupId>
      <artifactId>junit</artifactId>
      <scope>test</scope>
    </dependency>
  </dependencies>
  <build>
    <finalName>mqtt-server</finalName>
    <plugins>  
      <plugin>  
        <groupId>org.springframework.boot</groupId>  
        <artifactId>spring-boot-maven-plugin</artifactId>  
      </plugin>  
    </plugins> 
  </build>
</project>

新建springboot启动类Application.java

@SpringBootApplication
@RestController
public class Application {

    @RequestMapping("/")
    public String greeting() {
        return "welcome, mqtt server!";
    }

    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }
}

新建mqtt推送服务代码MqttServer.java

@Component
@ConfigurationProperties(prefix = "my")  // 配置文件中的前缀
public class MqttServer {
    private String hostName;   //地址
    private String clientId;   //clientId
    private String topicName;  //主题
    private MqttClient client ;  

    public MqttServer(){
        try {
            //获取客户端连接
            client=new MqttClient(hostName,clientId);

            //设置参数
            MqttConnectOptions options = new MqttConnectOptions();  
            options.setCleanSession(false);  
//          options.setUserName(userName);  
//          options.setPassword(passWord.toCharArray());  
            // 设置超时时间  
            //options.setConnectionTimeout(10);  
            // 设置会话心跳时间  
            options.setAutomaticReconnect(true);
            options.setKeepAliveInterval(20);  
            client.setCallback(new PushCallback());  
            client.connect(options);  
        } catch (MqttException e) {
            e.printStackTrace();
        } 
    }

    /***
     * 发送消息
     * @param sendMsg    消息内容
     */
    public void publish(String sendMsg){     
        try {     
            MqttTopic topic = client.getTopic(topicName);     
            System.out.println("发送的消息内容为:"+sendMsg);  
            MqttMessage message = new MqttMessage(sendMsg.getBytes("utf-8"));     
            message.setQos(1);     
            topic.publish(message);   
        } catch (Exception e) {     
            e.printStackTrace();     
        }     
    }

    public String getHostName() {
        return hostName;
    }

    public void setHostName(String hostName) {
        this.hostName = hostName;
    }

    public String getTopicName() {
        return topicName;
    }

    public void setTopicName(String topicName) {
        this.topicName = topicName;
    }

    public String getClientId() {
        return clientId;
    }

    public void setClientId(String clientId) {
        this.clientId = clientId;
    }

    public MqttClient getClient() {
        return client;
    }

    public void setClient(MqttClient client) {
        this.client = client;
    }
}

/***
 * 回调函数
 * @author ouyjm
 *
 */
class PushCallback implements MqttCallback {  
    @Override
    public void connectionLost(Throwable cause) {  
        // 连接丢失后，一般在这里面进行重连  
        System.out.println("连接断开，可以做重连");  
    }  

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
         // publish后会执行到这里  
        System.out.println("deliveryComplete---------"+ token.isComplete());  

    }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        // subscribe后得到的消息会执行到这里面  
        System.out.println("接收消息主题:"+topic);  
        System.out.println("接收消息Qos:"+message.getQos());  
        System.out.println("接收消息内容:"+new String(message.getPayload()));  

    }  
}  

新建接收推送消息的API接口MqttApi.java。

@RestController
@RequestMapping("/mqttServer")
public class MqttApi {
    @Autowired
    private MqttServer mqttServer;

    @RequestMapping(value = "/send", method = RequestMethod.GET)
    public String sendMessage(String message){
        mqttServer.publish(message);
        return "send \""+message+"\" to client success!";
    }

}

配置文件application.yml中修改了默认端口，以及mqtt服务器相关配置。如下：

server:
  port: 9090
my:
  hostName: tcp://localhost:1883
  clientId: "ibm-mqisdp"
  topicName: mqtt

以上为maven工程的代码。

---

使用方式1：
不考虑客户端实现调试。
- 启动mqtt服务，见上一章。
- 服务器上启动一个对mqtt topic的订阅，见上一章。
- 访问http://location:9090/mqttServer/send?message=8945782
即可在订阅的控制台看到8945782的输出。

使用方式2：
Android机器联调。
可以到IBM网站上下载Android的demo。
填入ClientID（通过查看端口占用状态可以知道lsof -i tcp:1883)、
Host（需要加上tcp://）、
Port（默认1883）后连接。
然后填入Topic订阅。
这时通过访问API接口，就能在手机上收到提醒了。