MqttConsumerConfig.java
3.8 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
package com.skua.modules.remotemeeting.mqtt;
import com.skua.modules.remotemeeting.service.IRemoteMeetingService;
import com.skua.tool.util.UniqIdUtils;
import lombok.extern.slf4j.Slf4j;
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.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
@Configuration
@Service
@Slf4j
public class MqttConsumerConfig {
@Value("${ar-remote-meeting.mqtt-url}")
private String hostUrl;
@Autowired
IRemoteMeetingService remoteMeetingService;
/**
* 客户端对象
*/
private MqttClient client;
/**
* 在bean初始化后连接到服务器
*/
@PostConstruct
public void init(){
connect();
}
/**
* 客户端连接服务端
*/
public void connect(){
try {
// clientId保证在多个连接中唯一,不然出现java.io.EOFException: null (a client identifier that is unique on the server being connected to)
String clientId = "jk_mqtt_client_5af98f9b" + UniqIdUtils.getInstance().getUniqID();
//创建MQTT客户端对象
client = new MqttClient(hostUrl,clientId,new MemoryPersistence());
//连接设置
MqttConnectOptions options = new MqttConnectOptions();
//是否清空session,设置为false表示服务器会保留客户端的连接记录,客户端重连之后能获取到服务器在客户端断开连接期间推送的消息
//设置为true表示每次连接到服务端都是以新的身份
options.setCleanSession(false);
//设置连接用户名
// options.setUserName(username);
//设置连接密码
// options.setPassword(password.toCharArray());
//设置超时时间,单位为秒
options.setConnectionTimeout(100);
//设置心跳时间 单位为秒,表示服务器每隔1.5*20秒的时间向客户端发送心跳判断客户端是否在线
options.setKeepAliveInterval(20);
//断开后自动重连
options.setAutomaticReconnect(true);
//设置遗嘱消息的话题,若客户端和服务器之间的连接意外断开,服务器将发布客户端的遗嘱信息
options.setWill("willTopic",(clientId + "与服务器断开连接").getBytes(),0,false);
//设置回调
client.setCallback(new MqttConsumerCallBack(this.remoteMeetingService, client));
client.connect(options);
//订阅主题
//消息等级,和主题数组一一对应,服务端将按照指定等级给订阅了主题的客户端推送消息
int[] qos = {1,1};
//主题
String[] topics = {"room/#","topic2"};
//订阅主题
client.subscribe(topics,qos);
log.info("第三方远程AR会议:mqtt连接成功!!!!!!!");
} catch (MqttException e) {
e.printStackTrace();
}
}
/**
* 主动断开与服务器之间的连接
*/
public void disConnect(){
try {
client.disconnect();
} catch (MqttException e) {
e.printStackTrace();
}
}
/**
* 订阅主题
*/
public void subscribe(String topic,int qos){
try {
client.subscribe(topic,qos);
} catch (MqttException e) {
e.printStackTrace();
}
}
}