android mqtt

水深无声 2022-06-03 07:44 138阅读 0赞

package com.example.com.mqtt.simple.chat;
    
    import android.app.Activity;
    import android.os.Bundle;
    import android.os.CountDownTimer;
    import android.os.Handler;
    import android.os.Message;
    import android.util.Log;
    import android.view.View;
    import android.view.View.OnClickListener;
    import android.widget.Button;
    import android.widget.EditText;
    import android.widget.TextView;
    
    import com.example.com.mqtt.simple.utils.MqttClientService;
    import com.ibm.micro.client.mqttv3.MqttException;
    import com.ibm.micro.client.mqttv3.MqttPersistenceException;
    import com.ibm.micro.client.mqttv3.MqttSecurityException;
    
    public class MainActivity extends Activity {
        
        private EditText message_et;//消息内容
        private EditText password_et;//密码
        private EditText user_name_et;//用户名
        private EditText topic_et;//主题
        private EditText host_name_et;//mqtt地址
        private EditText receive_et;//接收内容
        private EditText client_id_et;//客户id
        private EditText subscribe_et;//订阅
        private TextView state_tv;//状态
    
        private Button connect_bt;//连接
        private Button submit_bt;//发送
        private Button subscribe_bt;
    
        private MqttClientService service;//mqtt服务
        private int failedCount=0;//失败次数
        protected final int MAX_FAILED_COUNT = 3 ;//最高失败次数
        public Handler handler;//ui 处理
    
        @Override
        protected void onCreate(Bundle savedInstanceState) {
            super.onCreate(savedInstanceState);
            setContentView(R.layout.activity_main);
    
            initView();
            initViewValue();
            initHandler();
            //initMqtt();
            reload();
            initListen();
        }
        public void reload(){
            failedCount++;
            initMqtt();
            new CountDownTimer(5000, 1000) {
                @Override
                public void onTick(long arg0) {}
                @Override
                public void onFinish() {
                    if(failedCount>MAX_FAILED_COUNT ){
                        if(service!=null&&service.isConnect()){
                            return;
                        }
                        state_tv.setText("mqtt状态：未连接 ( "+failedCount+" 次重新连接失败！)");
                        return;
                    }
                    if(service==null || !service.isConnect()){
                        reload();
                    }
                }
            }.start();
        }
        private void initView(){
            state_tv = (TextView)findViewById(R.id.state_tv);
            connect_bt=(Button)findViewById(R.id.connect_bt);
            host_name_et=(EditText)findViewById(R.id.host_name_et);
            topic_et=(EditText)findViewById(R.id.topic_et);
            user_name_et=(EditText)findViewById(R.id.user_name_et);
            password_et=(EditText)findViewById(R.id.password_et);
            message_et=(EditText)findViewById(R.id.message_et);
            submit_bt=(Button)findViewById(R.id.submit_bt);
            receive_et=(EditText)findViewById(R.id.receive_et);
            client_id_et=(EditText)findViewById(R.id.client_id_et);
            subscribe_et=(EditText)findViewById(R.id.subscribe_et);
            subscribe_bt=(Button)findViewById(R.id.subscribe_bt);
        }
        private void initViewValue() {
            host_name_et.setText("tcp://192.168.0.107:61613");
            topic_et.setText("bxh");
            user_name_et.setText("admin");
            password_et.setText("password");
            message_et.setText("password");
            subscribe_et.setText("bxh");
        }
        private void initHandler() {
            handler = new Handler(){
                @Override
                public void handleMessage(Message msg) {
                    switch (msg.what) {
                    case 1:
                        state_tv.setText("mqtt状态：未连接");
                        connect_bt.setClickable(true);
                        failedCount=0;
                        reload();
                        break;
                    case 2:
                        //connect_bt.setClickable(false);
                        state_tv.setText("mqtt状态：已连接");
                        break;
                    case 3:
                        receive_et.setText(msg.getData().getString("msg"));
                        break;
                    case 4:
                        client_id_et.setText(msg.getData().getString("clientId"));
                        break;
                    default:
                        break;
                    }
    
                }
            };
        }
        private void initMqtt() {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        service = MqttClientService.getInstance(
                                host_name_et.getText().toString().trim(),
                                user_name_et.getText().toString().trim(),
                                password_et.getText().toString().trim(),
                                handler
                                );
                        service.subscribe("bxh");//订阅
    
                        Message msg =new Message();
                        msg.what=2;
                        handler.sendMessage(msg);//不出exception 连接成功
                        return;
                    } catch (MqttSecurityException e) {
                        Log.e("e", e.getMessage());
                    } catch (MqttException e) {
                        Log.e("e", e.getMessage());
                    }
                }
            }).start();
        }
        private void initListen() {
       //添加监听
            submit_bt.setOnClickListener(new OnClickListener() {
       //发送
    
                @Override
                public void onClick(View arg0) {
                    String topic = topic_et.getText().toString().trim();
                    String msg = message_et.getText().toString().trim();
                    if ("".equals(topic)||"".equals(msg)) {
                        return;
                    }
                    try {
                        service.publish(topic, msg);//推送
                    } catch (MqttPersistenceException e) {
                        Log.e("e", e.getMessage());
                    } catch (MqttException e) {
                        Log.e("e", e.getMessage());
                    }
                }
            });
    
            connect_bt.setOnClickListener(new OnClickListener() {
       //连接
    
                @Override
                public void onClick(View arg0) {
                    reload();
                }
            });
            subscribe_bt.setOnClickListener(new OnClickListener() {
       //订阅
    
                @Override
                public void onClick(View arg0) {
                    if (service!=null&&service.isConnect()) {
                        String topicName = subscribe_et.getText().toString().trim();
                        if(!"".equals(topicName)){
                            try {
                                service.subscribe(topicName);
                                subscribe_et.setText("");
                            } catch (MqttSecurityException e) {
                                Log.e("e", e.getMessage());
                            } catch (MqttException e) {
                                Log.e("e", e.getMessage());
                            }
                        }
                    }
                }
            });
    
        }
    }

package com.example.com.mqtt.simple.utils;
    
    import java.util.Arrays;
    import java.util.HashMap;
    import java.util.Map;
    
    import android.os.Bundle;
    import android.os.Handler;
    import android.os.Message;
    
    import com.ibm.micro.client.mqttv3.MqttCallback;
    import com.ibm.micro.client.mqttv3.MqttClient;
    import com.ibm.micro.client.mqttv3.MqttConnectOptions;
    import com.ibm.micro.client.mqttv3.MqttDeliveryToken;
    import com.ibm.micro.client.mqttv3.MqttException;
    import com.ibm.micro.client.mqttv3.MqttMessage;
    import com.ibm.micro.client.mqttv3.MqttPersistenceException;
    import com.ibm.micro.client.mqttv3.MqttSecurityException;
    import com.ibm.micro.client.mqttv3.MqttTopic;
    import com.ibm.micro.client.mqttv3.internal.MemoryPersistence;
    
    public class MqttClientService {
        
    
            private String host = "tcp://127.0.0.1:61613";
            private String clientid ; 
            private String userName = "admin";
            private String passWord = "password";
    
            private MqttClient client;  
            private MqttConnectOptions options; 
    
            private static MqttClientService instance;//单例
    
            private Map topicMap;//主题 map （去重复）
    
            private Handler handler;
    
            public void init() throws MqttSecurityException, MqttException {
                topicMap = new HashMap();
    
                clientid =System.currentTimeMillis()+"";//用户id 用当前时间
    
                client = new MqttClient(host, clientid, new MemoryPersistence());
    
                options = new MqttConnectOptions();  
                options.setUserName(userName);  
                options.setPassword(passWord.toCharArray()); 
                options.setCleanSession(true);
                options.setConnectionTimeout(10);  
                options.setKeepAliveInterval(20);
    
                client.setCallback(new PushCallback());
                //MqttTopic topic = client.getTopic("bxh");
                //options.setWill(topic, "Will".getBytes(), 2, true);
    
                client.connect(options);
    
                Message msg = new Message();
                Bundle data = new Bundle();
                data.putString("clientId", clientid);
                msg.setData(data);
                msg.what=4;
                handler.sendMessage(msg);//改变mainActivity 的cliendId
    
            }
    
            private MqttClientService(){}//单例
            private MqttClientService(String host,String userName,String password,Handler handler){
                this.host=host;
                this.userName=userName;
                this.passWord=password;
                this.handler=handler;
            }
            /**
             * 初始化service
             * @param host
             * @param userName
             * @param password
             * @param handler    ui 改变
             * @return
             * @throws MqttSecurityException
             * @throws MqttException
             */
            public static MqttClientService getInstance(String host,String userName,String password,Handler handler) throws MqttSecurityException, MqttException{
                if(instance==null){
       //单例
                    //instance = new MqttClientService();
                    instance = new MqttClientService(host,userName,password,handler);
                }
                instance.init();//连接服务器
                return instance;
            }
            /**
             * 消息推送
             * @param topicName  主题
             * @param message    消息
             * @throws MqttPersistenceException
             * @throws MqttException
             */
            public void publish(String topicName,String message) throws MqttPersistenceException, MqttException{
                client.getTopic(topicName).publish(new MqttMessage(message.getBytes()));
            }
            /**
             * 消息订阅
             * @param topicName    主题
             * @throws MqttSecurityException
             * @throws MqttException
             */
            public void subscribe(String topicName) throws MqttSecurityException, MqttException{
                topicMap.put(topicName,client.getTopic(topicName));//主题集合
    
                String[] topics= new String[topicMap.size()];
                int[] Qos  = new int[topicMap.size()];
    
                Arrays.fill(Qos, 1);
                topicMap.keySet().toArray(topics);
                client.subscribe(topics, Qos);
            }
            /**
             * 是否连接成功
             * @return
             */
            public boolean isConnect(){
                return client.isConnected();
            }
            /**
             * mqtt 回调
             * @author bxh
             *
             */
            class PushCallback implements MqttCallback {
                @Override
                public void connectionLost(Throwable cause) {
                    Message msg =new Message();
                    msg.what=1;
                    handler.sendMessage(msg);//通知失去连接
                }  
                @Override
                public void deliveryComplete(MqttDeliveryToken t) {
    
                }
                @Override
                public void messageArrived(MqttTopic topic, MqttMessage message) throws MqttException{
                    //System.out.println("接收消息主题 :" + topic);  
                    //System.out.println("接收消息 Qos:" + message.getQos());  
                    Message msg = new Message();
                    msg.what=3;
                    Bundle data = new Bundle();
                    data.putString("msg",  topic.getName()+": "+new String(message.getPayload()));
                    msg.setData(data);
                    handler.sendMessage(msg);//通知 消息到收到消息
                }  
            }
            /**
             * ui 改变
             * @param handler
             */
            public void setHandler(Handler handler){
                this.handler=handler;
            }
    }

<?xml version="1.0" encoding="utf-8"?>
    <manifest xmlns:android="http://schemas.android.com/apk/res/android"
        package="com.example.com.mqtt.simple.chat"
        android:versionCode="1"
        android:versionName="1.0" >
    
        <uses-sdk
            android:minSdkVersion="14"
            android:targetSdkVersion="21" />
    
    
    
    
        <application
            android:allowBackup="true"
            android:icon="@drawable/ic_launcher"
            android:label="@string/app_name"
            android:theme="@style/AppTheme" >
            <activity
                android:name=".MainActivity"
                android:label="@string/app_name" >
                <intent-filter>
                    <action android:name="android.intent.action.MAIN" />
    
                    <category android:name="android.intent.category.LAUNCHER" />
                </intent-filter>
            </activity>
        </application>
    
        <uses-permission android:name="android.permission.INTERNET"/>
        <uses-permission android:name="android.permission.WRITE_EXTERNAL_STORAGE"/>
    
    </manifest>