android客户端 Mqtt连接+定时发送+订阅消息

android客户端 Mqtt连接+定时发送+订阅消息 1、引入依赖包

//Android mqtt客户端
compile 'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.1.0'
compile 'org.eclipse.paho:org.eclipse.paho.android.service:1.1.1'

2、Mqtt service使用类

继承了service类，使用service订阅和发布消息

public class IotMqttService extends Service {
    private static MqttAndroidClient mqttAndroidClient;
    public final String TAG = IotMqttService.class.getSimpleName();
    public String HOST = "tcp:/
    public static void startService(Context mContext) {
        mContext.startService(new Intent(mContext, IotMqttService.class));
    }

    
    public static void publish(String topic, String message) {
        Integer qos = 2;
        Boolean retained = false;
        try {
            Log.v("IotMqttService：topic="+topic, "Mqtt发送消息："+message);
            //参数分别为：主题、消息的字节数组、服务质量、是否在服务器保留断开连接后的最后一条消息
            mqttAndroidClient.publish(topic, message.getBytes(), qos.intValue(), retained.booleanValue());
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    @Nullable
    @Override
    public IBinder onBind(Intent intent) {
        return null;
    }

    
    public void response(String topic, String result) {
        JSonObject jsonResult = JSONObject.parseObject(result);
        if(jsonResult.getString("equipmentNo").equals(WebSettingConfig.getInstance().getClientId())){
            if(topic.equals(QUERY_TOPIC)){ // 查询
                publish(RESPONSE_TOPIC,"{}");
            }else if(topic.equals(TURN_ON_TOPIC)){ // 开
                publish(RESPONSE_TOPIC,"{}");
            }else if(topic.equals(TURN_OFF_TOPIC)){ // 关
                publish(RESPONSE_TOPIC,"{}");
            }
        }
    }

    
    private void init() {
        String serverURI = HOST; //服务器地址（协议+地址+端口号）
        mqttAndroidClient = new MqttAndroidClient(this, serverURI, CLIENTID);
        mqttAndroidClient.setCallback(mqttCallback); //设置监听订阅消息的回调
        mMqttConnectOptions = new MqttConnectOptions();
        mMqttConnectOptions.setCleanSession(true); //设置是否清除缓存
        mMqttConnectOptions.setConnectionTimeout(10); //设置超时时间，单位：秒
        mMqttConnectOptions.setKeepAliveInterval(10); //设置心跳包发送间隔，单位：秒
        mMqttConnectOptions.setAutomaticReconnect(true); // 设置自动重连
        mMqttConnectOptions.setUserName(USERNAME); //设置用户名
        mMqttConnectOptions.setPassword(PASSWORD.toCharArray()); //设置密码

        doClientConnection();
    }

    
    public void MqttSubscribes(){
        try {
            //订阅开的主题、服务质量
            Log.v(TAG,"订阅topic："+TURN_ON_TOPIC);
            mqttAndroidClient.subscribe(TURN_ON_TOPIC, 2);
            //订阅关的主题、服务质量
            Log.v(TAG,"订阅topic："+TURN_OFF_TOPIC);
            mqttAndroidClient.subscribe(TURN_OFF_TOPIC, 2);
            //订阅查询的主题、服务质量
            Log.v(TAG,"订阅topic："+QUERY_TOPIC);
            mqttAndroidClient.subscribe(QUERY_TOPIC, 2);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    
    private void doClientConnection() {
        if (!mqttAndroidClient.isConnected() && isConnectIsNomarl()) {
            try {
                mqttAndroidClient.connect(mMqttConnectOptions, null, iMqttActionListener);
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }
    }

    
    private boolean isConnectIsNomarl() {
        ConnectivityManager connectivityManager = (ConnectivityManager) this.getApplicationContext().getSystemService(Context.CONNECTIVITY_SERVICE);
        NetworkInfo info = connectivityManager.getActiveNetworkInfo();
        if (info != null && info.isAvailable()) {
            String name = info.getTypeName();
            Log.i(TAG, "当前网络名称：" + name);
            return true;
        } else {
            Log.i(TAG, "没有可用网络");
            
            new Handler().postDelayed(new Runnable() {
                @Override
                public void run() {
                    doClientConnection();
                }
            }, 3000);
            return false;
        }
    }

    @Override
    public void onDestroy() {
        try {
            mqttAndroidClient.disconnect(); //断开连接
        } catch (MqttException e) {
            e.printStackTrace();
        }
        super.onDestroy();
    }
}

3、定时任务类

定时任务也是继承的service类

public class TimingService extends Service {
    private String TAG = "TimingService";
    int TIME_INTERVAL = 1000*60*1; // 设置3分钟执行一次
    PendingIntent pendingIntent;
    AlarmManager alarmManager;
    public static final String TEST_ACTION = "timingService";


    @Override
    public void onCreate() {
        super.onCreate();
        IntentFilter intentFilter = new IntentFilter(TEST_ACTION);
        registerReceiver(receiver, intentFilter);
        alarmManager = (AlarmManager) getSystemService(ALARM_SERVICE);
        Intent intent = new Intent();
        intent.setAction(TEST_ACTION);
        pendingIntent = PendingIntent.getBroadcast(this, 0, intent, 0);

        if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.M) {//6.0低电量模式需要使用该方法触发定时任务
            alarmManager.setExactAndAllowWhileIdle(AlarmManager.ELAPSED_REALTIME_WAKEUP, SystemClock.elapsedRealtime(), pendingIntent);
        } else if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.KITKAT) {//4.4以上 需要使用该方法精确执行时间
            alarmManager.setExact(AlarmManager.ELAPSED_REALTIME_WAKEUP, SystemClock.elapsedRealtime(), pendingIntent);
        } else {//4.4一下 使用老方法
            alarmManager.setRepeating(AlarmManager.ELAPSED_REALTIME_WAKEUP, SystemClock.elapsedRealtime(), TIME_INTERVAL, pendingIntent);
        }
    }

    @Override
    public void onDestroy() {
        super.onDestroy();
        unregisterReceiver(receiver);
    }

    @Nullable
    @Override
    public IBinder onBind(Intent intent) {
        return null;
    }

    BroadcastReceiver receiver = new BroadcastReceiver() {
        @Override
        public void onReceive(Context context, Intent intent) {
            String action = intent.getAction();
            if (TEST_ACTION.equals(action)) {
                Log.v(TAG,"定时任务执行");
                sendReadStatusOrder();
                if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.M) {
                    alarmManager.setExactAndAllowWhileIdle(AlarmManager.ELAPSED_REALTIME_WAKEUP, SystemClock.elapsedRealtime() + TIME_INTERVAL, pendingIntent);
                } else if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.KITKAT) {
                    alarmManager.setExact(AlarmManager.ELAPSED_REALTIME_WAKEUP, SystemClock.elapsedRealtime() + TIME_INTERVAL, pendingIntent);
                }
            }
        }
    };

    
    public void sendReadStatusOrder(){
        try{
            for (int i=3;i<11;i++){
                int timeNumber = 1000*(i-3);
                Message msg = new Message();
                msg.what = 0;
                msg.obj = "指令";
                sendHandler.sendMessageDelayed(msg,timeNumber);
            }
        }catch (Exception e){
            e.printStackTrace();
            Log.e(TAG,Log.getStackTraceString(e));
        }
    }

    Handler sendHandler = new Handler(){
        @Override
        public void handleMessage(Message msg) {
            String result = msg.obj.toString();

        }
    };
}

4、初始化service

在mainfest中注册

5、在activity中启动service以及发送消息

public class MainActivity extends AppCompatActivity {
    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);

        Intent mqttIntent = new Intent(MainActivity.this,IotMqttService.class);
        startService(mqttIntent);
        Intent timeIntent = new Intent(MainActivity.this,TimingService.class);
        startService(timeIntent);
    }

    public void sendMeessage(){
        IotMqttService.publish(IotMqttService.RESPONSE_TOPIC,paramsStr);
    }
}