YUSHENG/yusheng-android
8
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

修改名称。

Browse Source
This commit is contained in:
骆召亮
2025-11-07 09:22:39 +08:00
parent d9cf55b053
commit a8dcfbb6a7
2203 changed files with 3 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
BaseModule/src/main/java/com/xscm/moduleutil/service/EmqttState.java Normal file
View File

@@ -0,0 +1,26 @@
package com.xscm.moduleutil.service;
public enum EmqttState {
DISCONNECT(0), //断开连接
CONNECTED(1),//已连接
TOPICS_SUCCESS(2),//订阅成功
TOPICS_FAIL(3),//订阅失败
CANCEL_TOPICS_SUCCESS(4),//取消订阅成功
CANCEL_TOPICS_FAIL(5),//取消订阅失败
CLOSE_SUCCESS(6),//关闭成功
CLOSE_FAIL(7),
;//关闭失败
private int value;
EmqttState(int value) {
this.value = value;
}
public int getValue() {
return value;
}
}

23
BaseModule/src/main/java/com/xscm/moduleutil/service/HandlerUtil.java Normal file
View File

@@ -0,0 +1,23 @@
package com.xscm.moduleutil.service;
import android.content.Context;
import android.os.Handler;
import java.lang.ref.WeakReference;
public class HandlerUtil extends Handler {
private static HandlerUtil instance = null;
WeakReference<Context> mActivityReference;
public static HandlerUtil getInstance(Context context) {
if (instance == null) {
instance = new HandlerUtil(context.getApplicationContext());
}
return instance;
}
HandlerUtil(Context context) {
mActivityReference = new WeakReference<>(context);
}
}

124
BaseModule/src/main/java/com/xscm/moduleutil/service/IMConnectionService.java Normal file
View File

@@ -0,0 +1,124 @@
package com.xscm.moduleutil.service;
import android.app.Notification;
import android.app.NotificationChannel;
import android.app.NotificationManager;
import android.app.Service;
import android.content.Intent;
import android.os.Build;
import android.os.IBinder;
import android.util.Log;
import androidx.annotation.Nullable;
import com.blankj.utilcode.util.LogUtils;
import com.blankj.utilcode.util.ToastUtils;
import com.tencent.imsdk.v2.V2TIMManager;
import com.tencent.imsdk.v2.V2TIMSDKListener;
import com.tencent.imsdk.v2.V2TIMUserFullInfo;
import com.xscm.moduleutil.R;
import com.xscm.moduleutil.base.CommonAppContext;
import com.xscm.moduleutil.http.RetrofitClient;
public class IMConnectionService extends Service {
private static final String TAG = "IMConnectionService";
private static final int NOTIFICATION_ID = 2;
private static final String CHANNEL_ID = "im_connection_channel";
private final V2TIMSDKListener imSdkListener = new V2TIMSDKListener() {
@Override
public void onConnecting() {
Log.d(TAG, "IM connecting...");
}
@Override
public void onConnectSuccess() {//重连成功
Log.d(TAG, "IM connect success");
if (CommonAppContext.getInstance().playId != null) {
LogUtils.e("@@@", ""+CommonAppContext.getInstance().playId);
RetrofitClient.getInstance().roomUserReconnect(CommonAppContext.getInstance().playId);
}
}
@Override
public void onConnectFailed(int code, String error) {
Log.e(TAG, "IM connect failed, code: " + code + ", error: " + error);
}
@Override
public void onKickedOffline() {
Log.w(TAG, "IM kicked offline");
if (CommonAppContext.getInstance().playId != null) {
ToastUtils.showShort("您的账号已被挤下线");
try {
CommonAppContext.getInstance().clearLoginInfo();
} catch (ClassNotFoundException e) {
throw new RuntimeException(e);
}
}
}
@Override
public void onUserSigExpired() {
Log.w(TAG, "IM user sig expired");
}
@Override
public void onSelfInfoUpdated(V2TIMUserFullInfo info) {
Log.d(TAG, "IM self info updated");
}
};
@Override
public void onCreate() {
super.onCreate();
startForegroundService();
V2TIMManager.getInstance().addIMSDKListener(imSdkListener);
Log.d(TAG, "IMConnectionService created and listener registered");
}
@Override
public int onStartCommand(Intent intent, int flags, int startId) {
return START_STICKY; // 服务被杀死后会自动重启
}
@Nullable
@Override
public IBinder onBind(Intent intent) {
return null;
}
private void startForegroundService() {
if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.O) {
// 创建通知渠道
NotificationChannel channel = new NotificationChannel(
CHANNEL_ID,
"IM Connection Service",
NotificationManager.IMPORTANCE_LOW
);
NotificationManager manager = getSystemService(NotificationManager.class);
if (manager != null) {
manager.createNotificationChannel(channel);
}
// 创建通知
Notification notification = new Notification.Builder(this, CHANNEL_ID)
.setContentTitle("IM连接服务")
.setContentText("保持IM连接活跃")
.setSmallIcon(R.mipmap.default_avatar)
.setOngoing(true)
.build();
startForeground(NOTIFICATION_ID, notification);
}
}
@Override
public void onDestroy() {
super.onDestroy();
V2TIMManager.getInstance().removeIMSDKListener(imSdkListener);
Log.d(TAG, "IMConnectionService destroyed and listener unregistered");
}
}

221
BaseModule/src/main/java/com/xscm/moduleutil/service/MqttConnect.java Normal file
View File

@@ -0,0 +1,221 @@
package com.xscm.moduleutil.service;
import android.content.Context;
import android.os.Handler;
import android.os.Looper;
import android.util.Log;
import android.widget.Toast;
import com.blankj.utilcode.util.LogUtils;
import com.google.android.gms.common.api.Api;
import com.hjq.toast.ToastUtils;
import com.xscm.moduleutil.utils.logger.DataLogger;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import java.util.ArrayList;
public class MqttConnect {
private String HOST;
private String Tag = "MQTT";
private String clientId = "";
private static MqttClient mqttClient = null;
private Context context;
// 订阅主题
public static String shutdown = "";
public static String update_app = "";
public static String qx_hour_ranking = "";
public static String qx_redpacket_arrive="";//红包飘屏的主题
Handler handler = new Handler(Looper.getMainLooper());
String[] topic;
int[] qos = {1,2,3,0,0,0,0,0,0,0,0,0,0}; // 消息质量
private static MqttConnect instance;
public MqttConnect(Context context, String host, String clientId) {
this.HOST = host;
this.context = context;
this.clientId = clientId;
// 这里是你自己需要订阅的主题
shutdown = "qx_room_topic"; // 关机
update_app = "qx_xunlehui"; // 发送更新APP
// qx_hour_ranking = "qx_hour_ranking";
qx_hour_ranking = "qx_hour_ranking";
qx_redpacket_arrive="qx_redpacket_arrive";
ArrayList<String> topicList = new ArrayList<>();
topicList.add(shutdown);
topicList.add(update_app);
topicList.add(qx_hour_ranking);
topicList.add(qx_redpacket_arrive);
topic = topicList.toArray(new String[0]);
}
/**
* 单列模式,只能实例化一次
* @param context
* @param host
* @param clientId
* @return
*/
public static synchronized MqttConnect getInstance(Context context, String host, String clientId) {
if (instance == null) {
instance = new MqttConnect(context, host, clientId);
}
return instance;
}
/**
* 客户端connect连接mqtt服务器
**/
public void mqttClient()
{
// close();
// handler.postDelayed(new Runnable() {
// @Override
// public void run() {
try {
// uiTip("MQTT开始连接");
MqttConnectOptions options = mqttConnectOptions();
mqttClient.setCallback(new MqttInitCallback(context, HOST, clientId));
mqttClient.connect(options);
// sub(topic,qos);
sub(shutdown);
sub(update_app);
sub(qx_hour_ranking);
sub(qx_redpacket_arrive);
// uiTip("MQTT连接成功");
}catch (MqttException e){
// uiTip("MQTT连接失败,准备重连。。。："+e.getMessage());
handler.postDelayed(new Runnable() {
@Override
public void run() {
Log.e(Tag,"开始重连。。。");
mqttClient();
}
},3000);
}
// }
// },200);
}
/**
* 在主线程弹出消息
* @param msg
*/
private void uiTip(String msg){
Log.d(Tag,msg);
handler.post(new Runnable() {
@Override
public void run() {
// Toast.makeText(context.getApplicationContext(), msg, Toast.LENGTH_SHORT).show();
LogUtils.e("mqtt","连接成功");
ToastUtils.show(msg);
}
});
}
/**
* MQTT连接参数设置
*/
private MqttConnectOptions mqttConnectOptions()
throws MqttException {
mqttClient = new MqttClient(HOST, clientId, new MemoryPersistence());
MqttConnectOptions options = new MqttConnectOptions();
options.setUserName("public");
options.setConnectionTimeout(10);
options.setCleanSession(true);
options.setConnectionTimeout(10);
options.setKeepAliveInterval(10);
return options;
}
/**
* 关闭MQTT连接
*/
public void close(){
if(mqttClient != null && mqttClient.isConnected()){
try {
mqttClient.close();
mqttClient.disconnect();
mqttClient = null;
} catch (MqttException e) {
Log.e(Tag,"关闭MQTT连接报错："+e.getMessage());
// ToastUtils.show("关闭MQTT连接报错");
}
}else {
Log.d(Tag,"Mqtt已关闭");
}
}
/**
* 向某个主题发布消息 默认qos：1
*/
public static void pub(String topic, String msg) throws MqttException {
MqttMessage mqttMessage = new MqttMessage();
mqttMessage.setPayload(msg.getBytes());
MqttTopic mqttTopic = mqttClient.getTopic(topic);
MqttDeliveryToken token = mqttTopic.publish(mqttMessage);
token.waitForCompletion();
}
/**
* 向某个主题发布消息
*
* @param topic: 发布的主题
* @param msg: 发布的消息
* @param qos: 消息质量 Qos：0、1、2
*/
public void pub(String topic, String msg, int qos) throws MqttException {
MqttMessage mqttMessage = new MqttMessage();
mqttMessage.setQos(qos);
mqttMessage.setPayload(msg.getBytes());
MqttTopic mqttTopic = mqttClient.getTopic(topic);
MqttDeliveryToken token = mqttTopic.publish(mqttMessage);
token.waitForCompletion();
}
/**
* 订阅某一个主题 ，此方法默认的的Qos等级为：1
*
* @param topic 主题
*/
public void sub(String topic){
try {
mqttClient.subscribe(topic);
} catch (MqttException e) {
Log.e(Tag,"MQTT主题订阅失败：" + e.getMessage());
// uiTip("MQTT主题订阅失败");
}
}
/**
* 订阅某一个主题，可携带Qos
*
* @param topic 所要订阅的主题
* @param qos
* 消息质量：0最多发送一次，不保证消息能够到达接收端，也不负责重发
* 1至少发送一次，确保消息能够到达接收端，但可能会导致消息重复
* 2确保消息恰好被接收一次
*/
public void sub(String[] topic, int[] qos){
try {
mqttClient.subscribe(topic, qos);
}catch (MqttException e){
Log.e(Tag,"订阅主题失败："+e.getMessage());
}
}
// 原文链接：https://blog.csdn.net/Fyx1987496919/article/details/140516525
}

234
BaseModule/src/main/java/com/xscm/moduleutil/service/MqttInitCallback.java Normal file
View File

@@ -0,0 +1,234 @@
package com.xscm.moduleutil.service;
import android.content.Context;
import android.os.Handler;
import android.os.Looper;
import android.util.Log;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.blankj.utilcode.util.GsonUtils;
import com.blankj.utilcode.util.LogUtils;
import com.hjq.toast.ToastUtils;
import com.orhanobut.logger.Logger;
import com.xscm.moduleutil.bean.MqttXlhEnd;
import com.xscm.moduleutil.bean.XLHBean;
import com.xscm.moduleutil.event.HourlyBean;
import com.xscm.moduleutil.event.RedBean;
import com.xscm.moduleutil.event.RoomGiftRunable;
import com.xscm.moduleutil.utils.SpUtil;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.greenrobot.eventbus.EventBus;
import java.util.List;
public class MqttInitCallback implements MqttCallback {
private String Tag = "MqttInitCallback";
private String HOST;
private String clientId = "";
private MqttConnect mqttConnect = null;
private Context context = null;
MqttInitCallback(Context context, String host, String clientId) {
this.context = context;
this.HOST = host;
this.clientId = clientId;
}
/**
* 连接丢失
*/
@Override
public void connectionLost(Throwable cause) {
Log.d(Tag, "mqtt连接断开,执行重连");
// ToastUtils.show("mqtt连接断开,执行重连");
mqttConnect = MqttConnect.getInstance(context, HOST, clientId);
mqttConnect.mqttClient();
}
/**
* subscribe订阅后得到的消息会执行到这里
*/
@Override
public void messageArrived(String topic, MqttMessage message) {
// Log.d(Tag,"topic");
// Log.d(Tag,topic);
String messageStr = message.toString();
Logger.e("MQTT", "收到的消息", "主题：" + topic + " 收到的消息：" + messageStr);
if (topic.equals("qx_room_topic")) {
// ToastUtils.show("收到礼物飘屏");
receiveMessage(topic, messageStr);
} else if (topic.equals("qx_xunlehui")) {
// ToastUtils.show("收到轮盘飘屏");
receiveXlhMessage(messageStr);
} else if (topic.equals("qx_hour_ranking")) {
receiveQXHourRanking(topic, messageStr);
} else if (topic.equals("qx_redpacket_arrive")) {
receiveRed(topic, messageStr);
}
}
private void receiveRed(String topic, String messageStr) {
try {
JSONObject jsonObject = JSON.parseObject(messageStr);
String message = jsonObject.getString("msg");
// 将事件处理放到主线程执行
new Handler(Looper.getMainLooper()).post(() -> {
processRedMessage(topic, message);
});
} catch (Exception e) {
Log.e("MQTT", "解析MQTT消息异常", e);
}
}
private void processRedMessage(String topic, String message) {
try {
// 如果 data 是集合字符串
// 解析为集合
RedBean dataList = JSON.parseObject(message, RedBean.class);
// 在主线程处理集合数据
new Handler(Looper.getMainLooper()).post(() -> {
processDataRed(dataList);
});
} catch (Exception e) {
Log.e("MQTT", "解析MQTT消息异常", e);
}
}
private void processDataRed(RedBean dataList) {
// 遍历集合并发送每个元素
// for (HourlyBean dataItem : dataList) {
// EventBus.getDefault().post(dataItem);
// }
// 或者发送整个集合
EventBus.getDefault().post(dataList);
}
private void receiveQXHourRanking(String topic, String data) {
try {
JSONObject jsonObject = JSON.parseObject(data);
int type = jsonObject.getIntValue("type");
String message = jsonObject.getString("msg");
// 将事件处理放到主线程执行
new Handler(Looper.getMainLooper()).post(() -> {
processMessage(topic, message);
});
} catch (Exception e) {
Log.e("MQTT", "解析MQTT消息异常", e);
// ToastUtils.show("收到礼物飘屏,解析异常");
}
}
private void processMessage(String topic, String data) {
try {
// 如果 data 是集合字符串
if (isJsonArray(data)) {
// 解析为集合
List<HourlyBean> dataList = JSON.parseArray(data, HourlyBean.class);
// 在主线程处理集合数据
new Handler(Looper.getMainLooper()).post(() -> {
processDataList(dataList);
});
}
} catch (Exception e) {
Log.e("MQTT", "解析MQTT消息异常", e);
}
}
// 处理集合数据
private void processDataList(List<HourlyBean> dataList) {
// 遍历集合并发送每个元素
// for (HourlyBean dataItem : dataList) {
// EventBus.getDefault().post(dataItem);
// }
// 或者发送整个集合
EventBus.getDefault().post(dataList);
}
// 判断是否为 JSON 数组
private boolean isJsonArray(String jsonString) {
try {
return JSON.parseArray(jsonString) != null;
} catch (Exception e) {
return false;
}
}
private void receiveMessage(String topic, String data) {
try {
JSONObject jsonObject = JSON.parseObject(data);
int type = jsonObject.getIntValue("type");
String message = jsonObject.getString("msg");
// 将事件处理放到主线程执行
new Handler(Looper.getMainLooper()).post(() -> {
processMessageType(type, message);
});
} catch (Exception e) {
Log.e("MQTT", "解析MQTT消息异常", e);
// ToastUtils.show("收到礼物飘屏,解析异常");
}
}
private void processMessageType(int type, String message) {
switch (type) {
case 5019://推送所有人-横幅礼物通知
new RoomGiftRunable(message).run();
break;
case 8000:
// XLHBean xlhBean= GsonUtils.fromJson(message, XLHBean.class);
// if (xlhBean!=null && xlhBean.getRoom_id()!=null && SpUtil.getMyRoomId()!=null) {
// if (xlhBean.getRoom_id().equals(SpUtil.getMyRoomId())) {
// if (xlhBean.getFrom_type()==3) {
LogUtils.e("MQTT", "收到消息" + message);
MqttXlhEnd mqttXlhEnd = new MqttXlhEnd();
mqttXlhEnd.setMessage(message);
EventBus.getDefault().post(mqttXlhEnd);
// }
// }
// }
break;
default:
break;
}
}
private void receiveXlhMessage(String messageStr) {
try {
JSONObject jsonObject = JSON.parseObject(messageStr);
int type = jsonObject.getIntValue("type");
String message = jsonObject.getString("msg");
XLHBean xlhBean = JSON.parseObject(message, XLHBean.class);
// 将事件处理放到主线程执行
new Handler(Looper.getMainLooper()).post(() -> {
processMessageType(type, message);
EventBus.getDefault().post(xlhBean);
});
} catch (Exception e) {
Log.e("MQTT", "解析MQTT消息异常", e);
// ToastUtils.show("收到轮盘飘屏,解析异常");
}
}
/**
* publish发布成功后会执行到这里
*/
@Override
public void deliveryComplete(IMqttDeliveryToken token) {
}
// 原文链接：https://blog.csdn.net/Fyx1987496919/article/details/140516525
}

9
BaseModule/src/main/java/com/xscm/moduleutil/service/MyEmqttConnectListener.java Normal file
View File

@@ -0,0 +1,9 @@
package com.xscm.moduleutil.service;
public interface MyEmqttConnectListener {
void onConnectSuccess();
void onConnectFailure();
}

7
BaseModule/src/main/java/com/xscm/moduleutil/service/MyEmqttMesgListener.java Normal file
View File

@@ -0,0 +1,7 @@
package com.xscm.moduleutil.service;
public interface MyEmqttMesgListener {
void messageArrived(String topic, String mesg);
}

8
BaseModule/src/main/java/com/xscm/moduleutil/service/MyEmqttSubscribeListener.java Normal file
View File

@@ -0,0 +1,8 @@
package com.xscm.moduleutil.service;
public interface MyEmqttSubscribeListener {
void onSubscribeSuccess(String topic);
void onSubscribeFailure();
}

545
BaseModule/src/main/java/com/xscm/moduleutil/service/MyMqttService.java Normal file
View File

@@ -0,0 +1,545 @@
package com.xscm.moduleutil.service;
import android.app.Notification;
import android.app.NotificationChannel;
import android.app.NotificationManager;
import android.app.Service;
import android.content.Context;
import android.content.Intent;
import android.net.ConnectivityManager;
import android.net.NetworkInfo;
import android.os.Build;
import android.os.Handler;
import android.os.HandlerThread;
import android.os.IBinder;
import android.os.Looper;
import android.util.Log;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.blankj.utilcode.util.LogUtils;
import com.orhanobut.logger.Logger;
import com.xscm.moduleutil.R;
import com.xscm.moduleutil.bean.XLHBean;
import com.xscm.moduleutil.event.MqttBean;
import com.xscm.moduleutil.event.RoomGiftRunable;
import org.eclipse.paho.android.service.MqttAndroidClient;
import org.eclipse.paho.client.mqttv3.IMqttActionListener;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.IMqttToken;
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.greenrobot.eventbus.EventBus;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
public class MyMqttService extends Service implements MyEmqttConnectListener, MyEmqttMesgListener, MyEmqttSubscribeListener {
private final static String TAG = "lxj";
private static int qos = 2;
// private static String HOST = "tcp://1.13.181.248";//测试
private static String HOST = "tcp://62.234.12.147";//正式
private static MqttAndroidClient mqttAndroidClient;
private MqttConnectOptions mMqttConnectOptions;
private static boolean b = true;
// 使用线程安全的集合存储监听器
private static final CopyOnWriteArrayList<MyEmqttMesgListener> messageListeners = new CopyOnWriteArrayList<>();
private static final CopyOnWriteArrayList<MyEmqttConnectListener> connectListeners = new CopyOnWriteArrayList<>();
private static final CopyOnWriteArrayList<MyEmqttSubscribeListener> subscribeListeners = new CopyOnWriteArrayList<>();
private static final String TOPIC_BOSS = "qx_room_topic";
private static final String TOPIC_XLH = "qx_xunlehui";
// 添加后台线程处理
private HandlerThread mqttHandlerThread;
private Handler mqttHandler;
private ExecutorService messageExecutorService;
// 添加连接重试限制
private static final int MAX_RETRY_ATTEMPTS = 5;
private int retryCount = 0;
// 服务状态
private static boolean isServiceRunning = false;
private static final int NOTIFICATION_ID = 1;
private static final String CHANNEL_ID = "mqtt_channel";
@Override
public void onCreate() {
super.onCreate();
isServiceRunning = true;
// 创建专用的HandlerThread处理MQTT操作
mqttHandlerThread = new HandlerThread("MqttServiceThread");
mqttHandlerThread.start();
mqttHandler = new Handler(mqttHandlerThread.getLooper());
// 创建线程池处理消息
messageExecutorService = Executors.newCachedThreadPool();
// 启动前台服务
startForegroundService();
try {
init();
} catch (MqttException e) {
Log.e(TAG, "MQTT初始化异常", e);
}
}
@Override
public int onStartCommand(Intent intent, int flags, int startId) {
return START_STICKY;
}
@Override
public IBinder onBind(Intent intent) {
return null;
}
private void startForegroundService() {
if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.O) {
// 创建通知渠道
NotificationChannel channel = new NotificationChannel(
CHANNEL_ID,
"MQTT Service",
NotificationManager.IMPORTANCE_LOW
);
NotificationManager manager = getSystemService(NotificationManager.class);
if (manager != null) {
manager.createNotificationChannel(channel);
}
// 创建通知
Notification notification = new Notification.Builder(this, CHANNEL_ID)
.setContentTitle("消息服务")
.setContentText("正在接收实时消息")
.setSmallIcon(R.mipmap.default_avatar)
.setOngoing(true)
.build();
startForeground(NOTIFICATION_ID, notification);
}
}
/**
* 初始化
*/
private void init() throws MqttException {
String CLIENTID = "android-" + MqttClient.generateClientId();
mqttAndroidClient = new MqttAndroidClient(this, HOST, CLIENTID);
mqttAndroidClient.setCallback(mqttCallback);
mMqttConnectOptions = new MqttConnectOptions();
mMqttConnectOptions.setCleanSession(true);
mMqttConnectOptions.setConnectionTimeout(10);
mMqttConnectOptions.setKeepAliveInterval(10);
mMqttConnectOptions.setUserName("public");
mMqttConnectOptions.setMqttVersion(MqttConnectOptions.MQTT_VERSION_3_1);
if (mqttAndroidClient != null && !mqttAndroidClient.isConnected()) {
doClientConnection();
}
}
private void doClientConnection() throws MqttException {
// 在后台线程执行连接操作
mqttHandler.post(() -> {
if (mqttAndroidClient != null && !mqttAndroidClient.isConnected() && isConnectIsNomarl() && b) {
try {
mqttAndroidClient.connect(mMqttConnectOptions, null, iMqttActionListener);
} catch (Exception e) {
Log.e(TAG, "MQTT连接异常", e);
handleConnectionFailure();
}
}
});
}
private void handleConnectionFailure() {
retryCount++;
if (retryCount < MAX_RETRY_ATTEMPTS) {
// 延迟重试
mqttHandler.postDelayed(() -> {
if (b) {
try {
doClientConnection();
} catch (MqttException e) {
Log.e(TAG, "MQTT连接异常", e);
}
}
}, 5000);
} else {
Log.w(TAG, "达到最大重试次数，停止重试");
retryCount = 0;
}
}
/**
* 判断网络是否连接
*/
private boolean isConnectIsNomarl() {
try {
ConnectivityManager connectivityManager = (ConnectivityManager) getSystemService(Context.CONNECTIVITY_SERVICE);
if (connectivityManager != null) {
NetworkInfo info = connectivityManager.getActiveNetworkInfo();
if (info != null && info.isAvailable()) {
String name = info.getTypeName();
Log.i(TAG, "当前网络名称：" + name);
return true;
} else {
Log.i(TAG, "没有可用网络");
return false;
}
}
} catch (Exception e) {
Log.e(TAG, "检查网络状态异常", e);
}
return false;
}
//MQTT是否连接成功的监听
private IMqttActionListener iMqttActionListener = new IMqttActionListener() {
@Override
public void onSuccess(IMqttToken arg0) {
retryCount = 0;
// 通知所有连接监听器
for (MyEmqttConnectListener listener : connectListeners) {
if (listener != null) {
listener.onConnectSuccess();
}
}
Logger.e(TAG, "链接状态：", "链接成功");
subscribe(TOPIC_BOSS);
subscribe(TOPIC_XLH);
}
@Override
public void onFailure(IMqttToken arg0, Throwable arg1) {
// 通知所有连接监听器
for (MyEmqttConnectListener listener : connectListeners) {
if (listener != null) {
listener.onConnectFailure();
}
}
Logger.e(TAG, "链接状态：", "链接失败: " + arg1.getMessage());
// 在后台线程处理重连
mqttHandler.postDelayed(() -> {
if (b) {
handleConnectionFailure();
}
}, 1000);
}
};
//订阅主题的回调
private MqttCallback mqttCallback = new MqttCallback() {
@Override
public void messageArrived(String topic, MqttMessage message) throws Exception {
// 将消息处理放到后台线程执行
messageExecutorService.execute(() -> {
try {
String messageStr = message.toString();
Logger.e(TAG, "收到的消息", "主题：" + topic + " 收到的消息：" + messageStr);
if (topic.equals(TOPIC_BOSS)) {
receiveMessage(topic, messageStr);
} else if (topic.equals(TOPIC_XLH)) {
receiveXlhMessage(messageStr);
}
} catch (Exception e) {
Log.e(TAG, "处理MQTT消息异常", e);
}
});
}
@Override
public void deliveryComplete(IMqttDeliveryToken arg0) {
LogUtils.e("deliveryComplete---------");
}
@Override
public void connectionLost(Throwable arg0) {
// 通知所有连接监听器
for (MyEmqttConnectListener listener : connectListeners) {
if (listener != null) {
listener.onConnectFailure();
}
}
Logger.e(TAG, "链接状态：", "链接断开: " + arg0.getMessage());
// 在后台线程处理重连
mqttHandler.postDelayed(() -> {
if (b) {
try {
doClientConnection();
} catch (MqttException e) {
Log.e(TAG, "MQTT连接异常", e);
}
}
}, 1000);
}
};
private void receiveXlhMessage(String messageStr) {
try {
JSONObject jsonObject = JSON.parseObject(messageStr);
int type = jsonObject.getIntValue("type");
String message = jsonObject.getString("msg");
XLHBean xlhBean=JSON.parseObject(message, XLHBean.class);
// 将事件处理放到主线程执行
new Handler(Looper.getMainLooper()).post(() -> {
// processMessageType(type, message);
EventBus.getDefault().post(xlhBean);
});
} catch (Exception e) {
Log.e(TAG, "解析MQTT消息异常", e);
}
}
private void receiveMessage(String topic, String data) {
try {
JSONObject jsonObject = JSON.parseObject(data);
int type = jsonObject.getIntValue("type");
String message = jsonObject.getString("msg");
// 将事件处理放到主线程执行
new Handler(Looper.getMainLooper()).post(() -> {
processMessageType(type, message);
});
} catch (Exception e) {
Log.e(TAG, "解析MQTT消息异常", e);
}
}
private void processMessageType(int type, String message) {
switch (type) {
case 5019://推送所有人-横幅礼物通知
new RoomGiftRunable(message).run();
break;
}
}
/**
* 发布 （模拟其他客户端发布消息）
*/
public static void publish(String topic, String message) {
if (mqttAndroidClient == null) {
Logger.e(TAG, "mqttAndroidClient is null", "发送失败");
return;
}
// 在后台线程执行发布操作
new Handler(Looper.getMainLooper()).post(() -> {
try {
IMqttDeliveryToken publish = mqttAndroidClient.publish(topic, message.getBytes(), qos, false);
publish.setActionCallback(new IMqttActionListener() {
@Override
public void onSuccess(IMqttToken asyncActionToken) {
Logger.e(TAG, "发送消息", "发送成功");
}
@Override
public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
Logger.e(TAG, "发送消息", "发送失败: " + exception.getMessage());
}
});
} catch (Exception e) {
Logger.e(TAG, "发送消息", "发送异常: " + e.getMessage());
}
});
}
public static void subscribe(String topic) {
// 在后台线程执行订阅操作
new Handler(Looper.getMainLooper()).post(() -> {
try {
if (mqttAndroidClient != null && mqttAndroidClient.isConnected()) {
IMqttToken subToken = mqttAndroidClient.subscribe(topic, qos);
subToken.setActionCallback(new IMqttActionListener() {
@Override
public void onSuccess(IMqttToken asyncActionToken) {
for (MyEmqttSubscribeListener listener : subscribeListeners) {
if (listener != null) {
listener.onSubscribeSuccess(topic);
}
}
Logger.e(TAG, "订阅成功：" + topic);
}
@Override
public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
for (MyEmqttSubscribeListener listener : subscribeListeners) {
if (listener != null &&
!TOPIC_BOSS.equals(topic) &&
!TOPIC_XLH.equals(topic)) {
listener.onSubscribeFailure();
}
}
Logger.e(TAG, "订阅失败：" + topic + ", error: " + exception.getMessage());
}
});
}
} catch (Exception e) {
Log.e(TAG, "订阅异常：" + topic, e);
}
});
}
public static void cleanSubscribe(String topic) {
// 在后台线程执行取消订阅操作
new Handler(Looper.getMainLooper()).post(() -> {
try {
if (mqttAndroidClient != null && mqttAndroidClient.isConnected()) {
IMqttToken subToken = mqttAndroidClient.unsubscribe(topic);
subToken.setActionCallback(new IMqttActionListener() {
@Override
public void onSuccess(IMqttToken asyncActionToken) {
Logger.e(TAG, "取消成功" + topic);
}
@Override
public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
Logger.e(TAG, "取消失败" + topic + ", error: " + exception.getMessage());
}
});
}
} catch (Exception e) {
Log.e(TAG, "取消订阅异常：" + topic, e);
}
});
}
public static void closeConnection() {
new Handler(Looper.getMainLooper()).post(() -> {
try {
if (mqttAndroidClient != null && mqttAndroidClient.isConnected()) {
IMqttToken disconnect = mqttAndroidClient.disconnect();
disconnect.setActionCallback(new IMqttActionListener() {
@Override
public void onSuccess(IMqttToken asyncActionToken) {
Logger.e(TAG, "断开链接", "断开链接成功");
}
@Override
public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
Logger.e(TAG, "断开链接", "断开链接失败" + exception.getMessage());
}
});
}
} catch (Exception e) {
Log.e(TAG, "关闭连接异常", e);
}
});
}
/**
* 清理资源
*/
@Override
public void onDestroy() {
super.onDestroy();
b = false;
isServiceRunning = false;
try {
// 清理资源
if (messageExecutorService != null) {
messageExecutorService.shutdown();
try {
if (!messageExecutorService.awaitTermination(5, TimeUnit.SECONDS)) {
messageExecutorService.shutdownNow();
}
} catch (InterruptedException e) {
messageExecutorService.shutdownNow();
Thread.currentThread().interrupt();
}
}
if (mqttHandlerThread != null) {
mqttHandlerThread.quitSafely();
}
cleanSubscribe(TOPIC_BOSS);
cleanSubscribe(TOPIC_XLH);
if (mqttAndroidClient != null) {
mqttAndroidClient.disconnect();
mqttAndroidClient.unregisterResources();
mqttAndroidClient = null;
}
// 停止前台服务
stopForeground(true);
Logger.e(TAG, "服务关闭", "资源释放成功");
} catch (Exception e) {
Log.e(TAG, "服务关闭异常", e);
}
}
// 修改监听器管理方法
public static void addMyEmqttMesgListener(MyEmqttMesgListener myEmqttMesgListener) {
if (myEmqttMesgListener != null && !messageListeners.contains(myEmqttMesgListener)) {
messageListeners.add(myEmqttMesgListener);
}
}
public static void removeMyEmqttMesgListener(MyEmqttMesgListener myEmqttMesgListener) {
messageListeners.remove(myEmqttMesgListener);
}
public static void addMyEmqttConnectListener(MyEmqttConnectListener myEmqttConnectListener) {
if (myEmqttConnectListener != null && !connectListeners.contains(myEmqttConnectListener)) {
connectListeners.add(myEmqttConnectListener);
}
}
public static void removeMyEmqttConnectListener(MyEmqttConnectListener myEmqttConnectListener) {
connectListeners.remove(myEmqttConnectListener);
}
public static void addMyEmqttSubscribeListener(MyEmqttSubscribeListener myEmqttSubscribeListener) {
if (myEmqttSubscribeListener != null && !subscribeListeners.contains(myEmqttSubscribeListener)) {
subscribeListeners.add(myEmqttSubscribeListener);
}
}
public static void removeMyEmqttSubscribeListener(MyEmqttSubscribeListener myEmqttSubscribeListener) {
subscribeListeners.remove(myEmqttSubscribeListener);
}
@Override
public void onConnectSuccess() {
// 实现接口方法
}
@Override
public void onConnectFailure() {
// 实现接口方法
}
@Override
public void messageArrived(String topic, String mesg) {
LogUtils.e("lxj", "messageArrived:" + mesg);
}
@Override
public void onSubscribeSuccess(String topic) {
LogUtils.e("lxj", "onSubscribeSuccess:" + topic);
}
@Override
public void onSubscribeFailure() {
LogUtils.e("lxj", "onSubscribeFailure");
}
}

92
BaseModule/src/main/java/com/xscm/moduleutil/service/MyRoomSingleton.java Normal file
View File

@@ -0,0 +1,92 @@
package com.xscm.moduleutil.service;
import android.content.Context;
import android.os.Handler;
import android.os.Looper;
import com.xscm.moduleutil.bean.RoomSingleton;
import com.xscm.moduleutil.http.BaseObserver;
import com.xscm.moduleutil.http.RetrofitClient;
import io.reactivex.disposables.Disposable;
/**
* @Author lxj$
* @Time 2025-8-6 11:42:49$ $
* @Description 每日任务的定时任务单例模式$
*/
public class MyRoomSingleton {
private int count;
private static MyRoomSingleton instance;
private Handler handler;
private String taskId;
private boolean isTimerActive;
private Context context;
private MyRoomSingleton() {
this.context = context;
handler = new Handler(Looper.getMainLooper());
isTimerActive = false;
}
public static MyRoomSingleton getInstance() {
if (instance == null) {
instance = new MyRoomSingleton();
}
return instance;
}
public void startTimerTask(String taskId) {
this.taskId = taskId;
count = 0; // 重置计数
isTimerActive = true;
handler.postDelayed(timerRunnable, 60 * 1000); // 每分钟请求一次接口
}
private Runnable timerRunnable = new Runnable() {
@Override
public void run() {
if (isTimerActive && count < 5) { // 30分钟任务，每分钟请求一次，共30/60=0.5次，这里假设每分钟请求一次，共30次
fetchData();
handler.postDelayed(this, 60 * 1000);
} else {
stopTimerTask(); // 达到最大请求次数后停止定时任务
}
}
};
public void stopTimerTask() {
handler.removeCallbacksAndMessages(null); // 销毁定时器
isTimerActive = false;
}
public void onExitRoom() {
stopTimerTask(); // 退出房间时停止定时任务
}
public void onEnterRoom(String taskId) {
this.taskId = taskId;
if (count!=0 ){
startTimerTask(taskId); // 进入房间时启动定时任务，延迟为0表示立即执行
}else {
count = 0; // 重置请求计数
startTimerTask(taskId); // 进入房间时启动定时任务，延迟为0表示立即执行
}
}
private void fetchData() {
// 这里编写请求接口的代码，例如使用Retrofit或Volley等库
// 示例使用Retrofit:
RetrofitClient.getInstance().dailyTasksComplete(taskId, new BaseObserver<RoomSingleton>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(RoomSingleton roomSingleton) {
count++;
// 这里处理请求结果
if (roomSingleton.getIs_completed()==1){
// 任务完成，可以做一些后续操作
stopTimerTask();
}
}
});
}
}

24
BaseModule/src/main/java/com/xscm/moduleutil/service/RoomClearCardiacRunnable.java Normal file
View File

@@ -0,0 +1,24 @@
package com.xscm.moduleutil.service;
import com.xscm.moduleutil.bean.room.RoomClearCardiacAllModel;
import org.greenrobot.eventbus.EventBus;
public class RoomClearCardiacRunnable implements Runnable {
private String data;
public RoomClearCardiacRunnable(String data) {
this.data = data;
}
@Override
public void run() {
// RoomClearCardiacModel roomClearCardiacModel = JSON.parseObject(data, RoomClearCardiacModel.class);
// if (TextUtils.isEmpty(data)) {
EventBus.getDefault().post(new RoomClearCardiacAllModel(data));
// } else {
// EventBus.getDefault().post(new RoomClearCardiacAllModel());
// }
}
}