android端和javaWeb端实现Mqtt消息推送

环境说明：

服务器：apollo1.7 官网下载地址：http://activemq.apache.org/apollo/download.html

android的mqtt的Eclipse Paho 官网下载地址：http://www.eclipse.org/paho/downloads.php

****地址：https://download.****.net/download/chehec2010/10346796

安装服务器：

1、下载后解压，然后运行apache-apollo-1.6\bin\apollo create httbroker 创建服务器实例

android端和javaWeb端实现Mqtt消息推送

2、create mybroker之后会在bin目录下生成mybroker文件夹【文件夹下的文件内容解释】

一、users.properties：

用来配置可以使用服务器的用户以及相应的密码。
其在文件中的存储方式是：用户名=密码，如：

admin=password

表示新增一个用户，用户名是：admin，密码是：password

二、groups.properties：

持有群体的用户映射，可以通过组而不是单个用户简化访问控制列表。
可以为一个定义的组设置多个用户，用户之间用“|”隔开，如：
admins=admin|lily

表示admins组中有admin和lily两个用户

三、black-list.txt：
用来存放不允许连接服务器的IP地址，相当于黑名单类似的东西。
例如：10.20.9.147

表示上面IP不能够连接到服务器。

四、login.config：

是一个服务器认证的配置文件，为了安全apollo1.6版本提供了认证功能，只有相应的用户名和正确的密码才能够连接

服务器。

五、服务器主配置文件apollo.xml：
该配置文件用于控制打开的端口，队列，安全，虚拟主机设置等。
1、认证：可以使用<authenticationdomain="internal" />来配置是否需要连接认证，如果将其属性enable设置为false表示不用认证，任何人都可以连接服务器，默认为true

2、access_rule：可以在broker或者virtual_host中用于定义用户对服务器资源的各种行为。如：
<access_rule allow="users" action="connect create destroy send receive consume"/>表示群组users里面的用户可以对服务器资源进行的操作有：connect 、create、 destroy、 send 、receive 、consume。详细的操作说明见：
http://activemq.apache.org/apollo/documentation/user-manual.html
3、message stores：默认情况下apollo使用的是LevelDB store，但是推荐使用BDB store（跨平台的）只能够实用其中一种。使用LevelDB store的配置是：<leveldb_store directory="${apollo.base}/data"/>默认有提供不用任何修改。使用BDB store需要到网站下jar包支持http://download.oracle.com/maven/com/sleepycat/je/5.0.34/je-5.0.34.jar，将jar包放在服务器的lib目录下面，然后将配置文件改成：<bdb_store directory="${apollo.base}/data"/>即可。
4、connector：用于配置服务器支持的链接协议以及相应的端口。如：<connector id="tcp" bind="tcp://0.0.0.0:61613" connection_limit="2000" protocol="mqtt"/>表示支持tcp链接，使用的端口是61613，链接限制是2000，自动侦听的协议是mqtt协议。

android端和javaWeb端实现Mqtt消息推送

3、启动httbroker

配置参考如上说明。若非本机安装，修改apollo.xml文件中web_admin的ip地址，以便访问控制台。

<web_admin bind="http://0.0.0.0:61680"/>

<web_admin bind="https://0.0.0.0:61681"/>

启动服务:

./bin/apollo-broker-service start

脚本参数：apollo-broker-service {start|stop|restart|force-stop|status}

启动后，访问http://192.168.36.102:61680/或者https://192.168.36.102:61681。默认用户:admin / password

4、测试启动httbroker服务器

android端和javaWeb端实现Mqtt消息推送

5、访问服务器

输入：http://127.0.0.1:61680/

android端和javaWeb端实现Mqtt消息推送

说明apollo1.7服务器启动成功

下来编写android端代码环境androidstudio下

android代码：

package com.mqtt.seer.mqtttest;

import android.app.Activity;
import android.os.Bundle;
import android.os.Handler;
import android.os.Message;
import android.view.KeyEvent;
import android.widget.TextView;
import android.widget.Toast;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class MainActivity extends Activity {

    private TextView messageTv;

    private String host = "tcp://192.168.1.121:61613";
    private String userName = "admin";
    private String passWord = "password";

    private Handler handler;

    private MqttClient client;

    private String myTopic = "test/topic";

    private MqttConnectOptions options;

    private ScheduledExecutorService scheduler;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);

        messageTv = (TextView) findViewById(R.id.messagetv);

        init();

        handler = new Handler() {
            @Override
            public void handleMessage(Message msg) {
                super.handleMessage(msg);
                if(msg.what == 1) {
                    Toast.makeText(MainActivity.this, (String) msg.obj,
                            Toast.LENGTH_SHORT).show();
                    System.out.println("-----------------------------");
                } else if(msg.what == 2) {
                    Toast.makeText(MainActivity.this, "连接成功", Toast.LENGTH_SHORT).show();
                    try {
                        client.subscribe(myTopic, 1);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                } else if(msg.what == 3) {
                    Toast.makeText(MainActivity.this, "连接失败，系统正在重连", Toast.LENGTH_SHORT).show();
                }
            }
        };

        startReconnect();

    }

    private void startReconnect() {
        scheduler = Executors.newSingleThreadScheduledExecutor();
        scheduler.scheduleAtFixedRate(new Runnable() {

            @Override
            public void run() {
                if(!client.isConnected()) {
                    connect();
                }
            }
        }, 0 * 1000, 10 * 1000, TimeUnit.MILLISECONDS);
    }

    private void init() {
        try {
            //host为主机名，test为clientid即连接MQTT的客户端ID，一般以客户端唯一标识符表示，MemoryPersistence设置clientid的保存形式，默认为以内存保存
            client = new MqttClient(host, "test",
                    new MemoryPersistence());
            //MQTT的连接设置
            options = new MqttConnectOptions();
            //设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接
            options.setCleanSession(true);
            //设置连接的用户名
            options.setUserName(userName);
            //设置连接的密码
            options.setPassword(passWord.toCharArray());
            // 设置超时时间 单位为秒
            options.setConnectionTimeout(10);
            // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
            options.setKeepAliveInterval(20);
            //设置回调
            client.setCallback(new MqttCallback() {

                @Override
                public void connectionLost(Throwable cause) {
                    //连接丢失后，一般在这里面进行重连
                    System.out.println("connectionLost----------");
                }

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

                @Override
                public void messageArrived(String topicName, MqttMessage message)
                    throws Exception {
                    //subscribe后得到的消息会执行到这里面
                    System.out.println("messageArrived----------");
                    Message msg = new Message();
                    msg.what = 1;
                    msg.obj = topicName+"---"+message.toString();
                    handler.sendMessage(msg);
                }
            });
          connect();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private void connect() {
        new Thread(new Runnable() {

            @Override
            public void run() {
                try {
                    client.connect(options);
                    Message msg = new Message();
                    msg.what = 2;
                    handler.sendMessage(msg);
                } catch (Exception e) {
                    e.printStackTrace();
                    Message msg = new Message();
                    msg.what = 3;
                    handler.sendMessage(msg);
                }
            }
        }).start();
    }

    @Override
    public boolean onKeyDown(int keyCode, KeyEvent event) {
        if(client != null && keyCode == KeyEvent.KEYCODE_BACK) {
            try {
                client.disconnect();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        return super.onKeyDown(keyCode, event);
    }

    @Override
    protected void onDestroy() {
        super.onDestroy();
        try {
            scheduler.shutdown();
            client.disconnect();
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }
}

注意：

private String host = "tcp://192.168.1.121:61613";

填写本机的ip地址

java后端的代码：

package base;

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import javax.swing.*;
import java.awt.*;
import java.awt.event.ActionEvent;
import java.awt.event.ActionListener;

public class Mqtt extends JFrame{
    private static final long serialVersionUID = 1L;
    private JPanel panel;
    private JButton button;

    private MqttClient client;
    private String host = "tcp://192.168.1.121:61613";
    //  private String host = "tcp://localhost:1883";
    private String userName = "admin";
    private String passWord = "password";
    private MqttTopic topic;
    private MqttMessage message;

    private String myTopic = "test/topic";

    public Mqtt() {

        try {
            client = new MqttClient(host, "Server",
                    new MemoryPersistence());
            connect();
        } catch (Exception e) {
            e.printStackTrace();
        }

        Container container = this.getContentPane();
        panel = new JPanel();
        button = new JButton("发布话题");
        button.addActionListener(new ActionListener() {

            @Override
            public void actionPerformed(ActionEvent ae) {
                try {
                    MqttDeliveryToken token = topic.publish(message);
                    token.waitForCompletion();
                    System.out.println(token.isComplete()+"========");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
        panel.add(button);
        container.add(panel, "North");

    }

    private void connect() {

        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(false);
        options.setUserName(userName);
        options.setPassword(passWord.toCharArray());
        // 设置超时时间
        options.setConnectionTimeout(10);
        // 设置会话心跳时间
        options.setKeepAliveInterval(20);
        try {
            client.setCallback(new MqttCallback() {

                @Override
                public void connectionLost(Throwable cause) {
                    System.out.println("connectionLost-----------");
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {
                    System.out.println("deliveryComplete---------"+token.isComplete());
                }

                @Override
                public void messageArrived(String topic, MqttMessage arg1)
                        throws Exception {
                    System.out.println("messageArrived----------");

                }
            });

            topic = client.getTopic(myTopic);

            message = new MqttMessage();
            message.setQos(1);
            message.setRetained(true);
            System.out.println(message.isRetained()+"------ratained状态");
            message.setPayload("eeeeeaaaaaawwwwww---".getBytes());

            client.connect(options);
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    public static void main(String[] args) {
        Mqtt s = new Mqtt();
        s.setDefaultCloseOperation(JFrame.EXIT_ON_CLOSE);
        s.setSize(600, 370);
        s.setLocationRelativeTo(null);
        s.setVisible(true);
    }
}

注意这里也填写本地ip：

private String host = "tcp://192.168.1.121:61613";

当javaweb端和android端都连接httbroker的时候，java后端点击订阅，android端就可以收到消息

apollo文件介绍：

android端和javaWeb端实现Mqtt消息推送

注意：connector是订阅和发布消息的端口

主要参考：https://blog.****.net/join_wangzi/article/details/7779709

https://blog.****.net/djun100/article/details/25752491

https://blog.****.net/qq_29196551/article/details/51488340

自己动手敲了一遍，确实可以实现，依次作为笔录。感谢指点和批评!

demo的下载地址：https://download.****.net/download/chehec2010/10346818

android端和javaWeb端实现Mqtt消息推送

相关推荐