MQTT协议实现Android聊天功能

Android客户端实现消息推送无外乎以下几种方式:

1、轮询:客户端通过定时机制往服务端获取消息,这种方式太傻逼,不算真正的推送,费流量。

2、客户端和服务端建立长久连接:这种方式才是真正的推送,由服务端通过连接的通道主动推送给客户端,常用的有xmpp和mqtt协议。

MQTT协议

MQTT协议是由IBM提出的基于发布/订阅模型的消息传输协议,相比于XMPP,它显得非常轻量小巧,协议内容包括固定头部+可变头部+消息体,最下的情况下头部只需要两个字节,在传输开销上有着巨大的优势,可以节省流量和电量。

MQTT可以保证消息的可靠性,它包括三种不同的服务质量(最多只传一次、最少被传一次、一次且只传一次),如果客户端意外掉线,可以使用“遗愿”发布一条消息,同时支持持久订阅。

“至多一次” 

消息根据底层因特网协议网络尽最大努力进行传递。 可能会丢失消息。

例如,将此服务质量与通信环境传感器数据一起使用。 对于是否丢失个别读取或是否稍后立即发布新的读取并不重要。

“至少一次” 

保证消息抵达,但可能会出现重复。

“刚好一次” 

确保只收到一次消息

MQTT服务器搭建

这个请自行百度,这里我们不加以详述。

MQTT Android客户端实现

在使用之前,配置android端的mqtt需要下面这些参数:

Topic:订阅的事件。在图一中,也就是“主题”

URI:MQTT服务器的地址。

username & password:账户与密码

ClientId:客户端的ID,可以自定义,必须保证唯一性,否则连接服务器的时候会导致服务器断开

1、添加依赖

implementation 'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.1.0'

implementation 'org.eclipse.paho:org.eclipse.paho.android.service:1.1.1'

2、配置客户端相关参数

    public static final String MQTT_IP="192.168.0.207";//服务器地址

    public static final String MQTT_PORT="1883";//服务器端口

    public static final String MQTT_TOPIC="mqtt";//订阅主题

    public static final String MQTT_USERNAME="Test11";//用户名

    public static final String MQTT_USERID="12";//用户id

3、连接mqtt服务器

public class MqttConfig {

    private String TAG = "mqtt";

    /**MQTT配置参数**/

    private static String host = Constants.MQTT_IP;

    private static String port = Constants.MQTT_PORT;

    private static String userID = "";

    private static String passWord = "";

    private static String clientID = MqttAppState.getInstance().getIMEI();

    private int mqttRetryCount=0;

    /**MQTT状态信息**/

    private boolean isConnect = false;

    /**MQTT支持类**/

    private MqttAsyncClient mqttClient=null;

    private MqttListener mMqttListener;//mqtt回调函数

    private Handler mHandler=new Handler(new Handler.Callback() {

        @Override

        public boolean handleMessage(Message message) {

            switch (message.arg1) {

                case MqttTag.MQTT_STATE_CONNECTED:

                    if (BuildConfig.DEBUG) Log.d(TAG, "handleMessage: connected");

                    mMqttListener.onConnected();

                    mqttRetryCount=0;

                    break;

                case MqttTag.MQTT_STATE_FAIL:

                    if (BuildConfig.DEBUG) Log.d(TAG, "handleMessage: fail");

                    mMqttListener.onFail();

                    break;

                case MqttTag.MQTT_STATE_LOST:

                    if (BuildConfig.DEBUG) Log.d(TAG, "handleMessage: lost");

                    mMqttListener.onLost();

                    break;

                case MqttTag.MQTT_STATE_RECEIVE:

                    if (BuildConfig.DEBUG) Log.d(TAG, "handleMessage: receive");

                    MqttObject object= (MqttObject) message.obj;

                    mMqttListener.onReceive(object.getTopic(),object.getMessage());

                    break;

                case MqttTag.MQTT_STATE_SEND_SUCC:

                    if (BuildConfig.DEBUG) Log.d(TAG, "handleMessage: send");

                    mMqttListener.onSendSucc();

                    break;

            }

            return true;

        }

    });

    /**

    * 自带的监听类,判断Mqtt活动变化

    */

    private IMqttActionListener mIMqttActionListener=new IMqttActionListener() {

        @Override

        public void onSuccess(IMqttToken asyncActionToken) {

            isConnect=true;

            Message msg=new Message();

            msg.arg1=MqttTag.MQTT_STATE_CONNECTED;

            mHandler.sendMessage(msg);

        }

        @Override

        public void onFailure(IMqttToken asyncActionToken, Throwable exception) {

            isConnect=false;

            Message msg=new Message();

            msg.arg1=MqttTag.MQTT_STATE_FAIL;

            mHandler.sendMessage(msg);

        }

    };

    /**

    * 自带的监听回传类

    */

    private MqttCallback mMqttCallback=new MqttCallback() {

        @Override

        public void connectionLost(Throwable cause) {

            isConnect=false;

            Message msg=new Message();

            msg.arg1=MqttTag.MQTT_STATE_LOST;

            mHandler.sendMessage(msg);

        }

        @Override

        public void messageArrived(String topic, MqttMessage message) throws Exception {

            Message msg=new Message();

            msg.arg1=MqttTag.MQTT_STATE_RECEIVE;

            String content=new String(message.getPayload());

            MqttObject object=new MqttObject();

            object.setTopic(topic);

            object.setMessage(content);

            msg.obj=object;

            mHandler.sendMessage(msg);

        }

        @Override

        public void deliveryComplete(IMqttDeliveryToken token) {

            Message msg=new Message();

            msg.arg1=MqttTag.MQTT_STATE_SEND_SUCC;

            mHandler.sendMessage(msg);

        }

    };

    public MqttConfig(MqttListener lis){

        mMqttListener=lis;

    }

    public static void setMqttSetting(String host, String port, String userID, String passWord, String clientID){

        MqttConfig.host = host;

        MqttConfig.port = port;

        MqttConfig.userID = userID;

        MqttConfig.passWord = passWord;

        MqttConfig.clientID = clientID;

    }

    /**

    * 进行Mqtt连接

    */

    public void connectMqtt(){

        try {

            mqttClient=new MqttAsyncClient("tcp://"+this.host+":"+this.port ,

                    "ClientID"+this.clientID, new MemoryPersistence());

            mqttClient.connect(getOptions(), null, mIMqttActionListener);

            mqttClient.setCallback(mMqttCallback);

        } catch (MqttException e) {

            Log.e("yaolinnan","e:",e);

//            e.printStackTrace();

        }

    }

    /**

    * 断开Mqtt连接重新连接

    */

    public void reStartMqtt(){

        if(mqttRetryCount<5) {

            disConnectMqtt();

            connectMqtt();

            mqttRetryCount++;

        }else {

            Log.i(TAG,"mqtt server reconnect error!");

        }

    }

    /**

    * 断开Mqtt连接

    */

    public void disConnectMqtt(){

        try {

            mqttClient.disconnect();

            mqttClient = null;

            isConnect = false;

        } catch (MqttException e) {

            e.printStackTrace();

        }

    }

    /**

    * 向Mqtt服务器发送数据

    */

    public void pubMsg(String Topic, String Msg, int Qos){

        if(!isConnect){

            Log.d(TAG,"Mqtt连接未打开");

            return;

        }

        try {

            /** Topic,Msg,Qos,Retained**/

            mqttClient.publish(Topic,Msg.getBytes(),Qos,false);

        } catch (MqttException e) {

            e.printStackTrace();

        }

    }

    /**

    * 向Mqtt服务器发送数据

    */

    public void pubMsg(String Topic, byte[] Msg, int Qos){

        if(!isConnect){

            Log.d(TAG,"Mqtt连接未打开");

            return;

        }

        try {

            /** Topic,Msg,Qos,Retained**/

            mqttClient.publish(Topic,Msg,Qos,false);

        } catch (MqttException e) {

            e.printStackTrace();

        }

    }

    /**

    * 向Mqtt服务器订阅某一个Topic

    */

    public void subTopic(String Topic, int Qos){

        if(!isConnect){

            Log.d(TAG,"Mqtt连接未打开");

            return;

        }

        try {

            mqttClient.subscribe(Topic,Qos);

        } catch (MqttException e) {

            e.printStackTrace();

        }

    }

    public void subTopic(String[] Topic, int[] Qos){

        if(!isConnect){

            Log.d(TAG,"Mqtt连接未打开");

            return;

        }

        try {

            mqttClient.subscribe(Topic,Qos);

        } catch (MqttException e) {

            e.printStackTrace();

        }

    }

    /**

    * 设置Mqtt的连接信息

    */

    private MqttConnectOptions getOptions(){

        MqttConnectOptions options = new MqttConnectOptions();

        options.setCleanSession(true);//重连不保持状态

        if(this.userID!=null&&this.userID.length()>0&&this.passWord!=null&&this.passWord.length()>0){

            options.setUserName(this.userID);//设置服务器账号密码

            options.setPassword(this.passWord.toCharArray());

        }

        options.setConnectionTimeout(10);//设置连接超时时间

        options.setKeepAliveInterval(20);//设置保持活动时间,超过时间没有消息收发将会触发ping消息确认

//        options.setAutomaticReconnect(true);

//        options.setCleanSession(false);

        return options;

    }

    public boolean isConnect() {

        return isConnect;

    }

    public static String getClientID() {

        return clientID;

    }

    class MqttObject{

        String topic;

        String message;

        public String getTopic() {

            return topic;

        }

        public void setTopic(String topic) {

            this.topic = topic;

        }

        public String getMessage() {

            return message;

        }

        public void setMessage(String message) {

            this.message = message;

        }

    }

}

4、通过service后台服务管理mqtt的连接、消息的接收和发送

public class MqttService extends Service implements MqttListener{

    private static final int MESSAGE_CHECK=0;

    private static MqttConfig mqttConfig;

    private static List<MqttListener> mMqttListenerList=new ArrayList<>();

    private CheckMqttThread myThread;

    private Timer timer=new Timer(true);

    private Handler mHandler=new Handler(){

        @Override

        public void handleMessage(Message msg) {

            if(msg.what==MESSAGE_CHECK){

                if(mqttConfig!=null&&!mqttConfig.isConnect()){

                    mqttConfig.connectMqtt();

                }

            }

        }

    };

    @Override

    public void onCreate() {

        super.onCreate();

        mqttConfig=new MqttConfig(this);

        mqttConfig.connectMqtt();

    }

    @Override

    public int onStartCommand(Intent intent, int flags, int startId) {

        if(myThread==null){

            myThread=new CheckMqttThread();//定时检查mqtt服务是否连接

            timer.scheduleAtFixedRate(myThread,2000,10000);

        }

        if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.O) {//启动前台服务

            startForeground(1, getNotification());

        }

        return Service.START_STICKY;

    }

    private Notification getNotification() {

        Notification notification=null;

        NotificationManager notificationManager = (NotificationManager) getSystemService(Context.NOTIFICATION_SERVICE);

        //设置Notification的ChannelID,否则不能正常显示

        if(Build.VERSION.SDK_INT >= Build.VERSION_CODES.O) {

            NotificationChannel mChannel = new NotificationChannel("mqtt", getResources().getString(R.string.app_name), NotificationManager.IMPORTANCE_HIGH);

            notificationManager.createNotificationChannel(mChannel);

            notification = new Notification.Builder(getApplicationContext(), "mqtt").build();

        }else {

            Notification.Builder builder = new Notification.Builder(this)

                    .setSmallIcon(R.mipmap.ic_launcher)

                    .setContentTitle(getResources().getString(R.string.app_name))

                    .setContentText("");

            notification = builder.build();

        }

        return notification;

    }

    private class CheckMqttThread  extends TimerTask {

        @Override

        public void run() {

            mHandler.sendEmptyMessage(MESSAGE_CHECK);

        }

    }

    @Nullable

    @Override

    public IBinder onBind(Intent intent) {

        return null;

    }

    public static void addMqttListener(MqttListener listener){

        if (!mMqttListenerList.contains(listener)) {

            mMqttListenerList.add(listener);

        }

    }

    public static void removeMqttListener(MqttListener listener){

        mMqttListenerList.remove(listener);

    }

    public static MqttConfig getMqttConfig(){

        return mqttConfig;

    }

    @Override

    public void onConnected() {

        Log.i("mqtt","mqttserver connnected");

        if(mqttConfig!=null){

            mqttConfig.subTopic(Constants.MQTT_TOPIC,0);

        }

        for (MqttListener mqttListener : mMqttListenerList) {

            mqttListener.onConnected();

        }

    }

    @Override

    public void onFail() {

        Log.i("mqtt","mqttserver fail");

        if(mqttConfig!=null) {

            mqttConfig.reStartMqtt();

        }

        for (MqttListener mqttListener : mMqttListenerList) {

            mqttListener.onFail();

        }

    }

    @Override

    public void onLost() {

        Log.i("mqtt","mqttserver lost");

        if(mqttConfig!=null) {

            mqttConfig.reStartMqtt();

        }

        for (MqttListener mqttListener : mMqttListenerList) {

            mqttListener.onLost();

        }

    }

    @Override

    public void onReceive(String topic, String message) {

        Log.i("mqtt","mqttserver receive message:"+message);

        for (MqttListener mqttListener : mMqttListenerList) {

            mqttListener.onReceive(topic,message);

        }

    }

    @Override

    public void onSendSucc() {

        Log.i("mqtt","mqttserver send success!");

        for (MqttListener mqttListener : mMqttListenerList) {

            mqttListener.onSendSucc();

        }

    }

}

5、编写聊天界面

public class MainActivity extends AppCompatActivity implements MqttListener{

    private RecyclerView mMessageRV;

    private EditText mMessageET;

    private Button mSendBtn;

    private Gson gson=new Gson();

    private List<MqttMessage> messageList=new ArrayList<>();

    private MqttAdapter mAdapter;

    @Override

    protected void onCreate(Bundle savedInstanceState) {

        super.onCreate(savedInstanceState);

        setContentView(R.layout.activity_main);

        MqttService.addMqttListener(this);

        Intent intent=new Intent(this, MqttService.class);

        if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.O) {

            startForegroundService(intent);

        }else {

            startService(intent);

        }

        mMessageRV=findViewById(R.id.swipe_target);

        mMessageET=findViewById(R.id.send_edit);

        mSendBtn=findViewById(R.id.send_button);

        mSendBtn.setOnClickListener(new View.OnClickListener() {

            @Override

            public void onClick(View v) {

                String message=mMessageET.getText().toString().trim();

                if(!TextUtils.isEmpty(message)){

                    MqttMessage mqttMessage=new MqttMessage();

                    mqttMessage.setUserid(Constants.MQTT_USERID);

                    mqttMessage.setMessage(message);

                    mqttMessage.setUsername(Constants.MQTT_USERNAME);

                    mqttMessage.setSendTime(new SimpleDateFormat("yyyy-MM-dd HH:mm").format(new Date()));

                    MqttService.getMqttConfig().pubMsg(Constants.MQTT_TOPIC,gson.toJson(mqttMessage),0);

                }else {

                    MqttAppState.getInstance().showToast("不可发送为空的消息");

                }

            }

        });

        LinearLayoutManager layoutManager = new LinearLayoutManager(this) {

            @Override

            public RecyclerView.LayoutParams generateDefaultLayoutParams() {

                return new RecyclerView.LayoutParams(ViewGroup.LayoutParams.MATCH_PARENT,

                        ViewGroup.LayoutParams.WRAP_CONTENT);

            }

        };

        layoutManager.setOrientation(LinearLayoutManager.VERTICAL);

        mMessageRV.setLayoutManager(layoutManager);

        layoutManager.setReverseLayout(true);//倒序显示

        layoutManager.setStackFromEnd(true);

        mAdapter=new MqttAdapter(this,messageList);

        mMessageRV.setAdapter(mAdapter);

    }

    @Override

    protected void onDestroy() {

        super.onDestroy();

        MqttService.removeMqttListener(this);

    }

    @Override

    public void onConnected() {

    }

    @Override

    public void onFail() {

    }

    @Override

    public void onLost() {

    }

    @Override

    public void onReceive(String topic, String message) {

        if(topic.equals(Constants.MQTT_TOPIC)){

            MqttMessage mqttMessage=gson.fromJson(message,MqttMessage.class);

            if(messageList.size()>0)

                messageList.add(0,mqttMessage);

            else

                messageList.add(mqttMessage);

            mAdapter.setList(messageList);

            mMessageRV.scrollToPosition(0);

        }

    }

    @Override

    public void onSendSucc() {

        MqttAppState.getInstance().showToast("消息发送成功");

        mMessageET.setText("");

        mMessageET.clearFocus();

    }

}

6、效果示意图


这样基本上就完成了通过mqtt协议聊天的功能!

完整代码:https://github.com/ylnJohn/MqttDemo

关注我的个人公众号【南的糊途】,获取更多技术资讯和相关资源

最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 205,236评论 6 478
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 87,867评论 2 381
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 151,715评论 0 340
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 54,899评论 1 278
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 63,895评论 5 368
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 48,733评论 1 283
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 38,085评论 3 399
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 36,722评论 0 258
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 43,025评论 1 300
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,696评论 2 323
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,816评论 1 333
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 33,447评论 4 322
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 39,057评论 3 307
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,009评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,254评论 1 260
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 45,204评论 2 352
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 42,561评论 2 343

推荐阅读更多精彩内容