我所做的
我尝试了什么
组态
<transportConnector name="mqtt+nio" uri="mqtt+nio://0.0.0.0:1883maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
我的Android代码:(如果还不够，请告诉我）
public boolean connectIfNecessary() throws MqttException, IOException {
synchronized (synchLock) {
MqttConnectOptions connectionOptions = new MqttConnectOptions();
connectionOptions.setCleanSession(false);
connectionOptions.setUserName(userName);
connectionOptions.setPassword(password.toCharArray());
if (mqttClient == null) {
stash = new MessageStash(applicationRoot + "/" + userName);
mqttClient = new MqttAsyncClient(
brokerUrl,
userName,
new MqttDefaultFilePersistence(applicationRoot + "/" + userName)
);
Log.d(TAG, "Broker URL: " + brokerUrl);
Log.d(TAG, "Connection clientId: " + userName);
Log.d(TAG, "Application path: " + applicationRoot);
mqttClient.setCallback(new MqttCallback() {
@Override
public void connectionLost(Throwable cause) {
Log.e(TAG, "Connection lost. Cause: " + cause.toString());
service.onConnectionLost();
notification.updateStatus(Notification.STATUS_DISCONNECTED);
}
@Override
public void messageArrived(String topic, MqttMessage message) throws Exception {
ConnectionBinder recipient = recipients.get(getTopicFromInbound(topic));
if (recipient != null)
recipient.onMessageReceived(message.toString());
Log.d(TAG, "Message " + message + " received");
}
@Override
public void deliveryComplete(IMqttDeliveryToken token) {
Log.d(TAG, "Message delivery complete");
}
});
}
if(mqttClient.isConnected()) // connection is already active - we can subscribe to the topic synchronously (see connect method)
return true;
if (connecting) // connecting was earlier initiated from a different thread - just let things take their course
return false;
connecting = true;
notification.updateStatus(Notification.STATUS_CONNECTING);
mqttClient.connect(connectionOptions, null, new IMqttActionListener() {
@Override
public void onSuccess(IMqttToken asyncActionToken) {
connecting = false;
Log.d(TAG, "connected");
for (MessageStash.Message message : stash.get()) {
try {
send(message.topic(), message.body());
message.commit();
} catch (IOException e) {
// we can safely ignore it here because this code is executed after the connection is restored
// so there will be no need to stash the message, but even the connection will be lost while
// resubmitting messages here there will be no need to worry - the message will remain stashed
// because message.commit will not be executed
}
}
for (Map.Entry<String, ConnectionBinder> binder : recipients.entrySet())
subscribe(binder.getKey());
notification.updateStatus(Notification.STATUS_CONNECTED);
}
@Override
public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
connecting = false;
// todo: onConnectionLost only on recoverable exceptions
Log.e(TAG, "Failed to connect :" + exception.toString());
service.onConnectionLost();
notification.updateStatus(Notification.STATUS_DISCONNECTED);
}
});
return false;
}
}
private void subscribe(String topic) {
final String inboundTopic = getInboundTopic(topic);
try {
mqttClient.subscribe(inboundTopic, QoS_EXACLY_ONCE,
null,
new IMqttActionListener() {
@Override
public void onSuccess(IMqttToken iMqttToken) {
Log.d(TAG, "Successfully subscribed to " + inboundTopic);
}
@Override
public void onFailure(IMqttToken iMqttToken, Throwable throwable) {
Log.e(TAG, "Subscribe to " + inboundTopic + " failed: " + throwable.toString());
}
});
} catch (MqttException e) {
logger.log(String.format("Exception subscribing to %s", inboundTopic), e);
}
}
public void unregisterSubscriber(String topic) {
String inboundTopic = getInboundTopic(topic);
if (mqttClient.isConnected())
try {
mqttClient.unsubscribe(inboundTopic);
} catch (MqttException e) {
logger.log(String.format("Exception unsubscribing from %s", inboundTopic), e);
}
recipients.remove(topic);
}
public void send(String topic, String message) throws IOException {
String outboundTopic = getOutboundTopic(topic);
try {
mqttClient.publish(outboundTopic, message.getBytes(), QoS_EXACLY_ONCE, true);
Log.d(TAG, "published to " + outboundTopic + " :" + message);
} catch (MqttException e) {
switch (e.getReasonCode()) {
// todo: double check this is the only recoverable failure
// it seems likely that REASON_CODE_CLIENT_DISCONNECTING should also be here
// I am not 100% sure, but I've seen a message 'Publish of blah failed ' with this reason code
case MqttException.REASON_CODE_CLIENT_DISCONNECTING:
case MqttException.REASON_CODE_CLIENT_NOT_CONNECTED:
stash.put(topic, message); // stash it for when the connection comes back online;
break;
default:
logger.log(String.format("Exception publishing to %s", outboundTopic), e);
break;
}
}
}
public static final String TAG = "MQTT Connection";
private static final int QoS_EXACLY_ONCE = 2;
private final Service service;
private final Notification notification;
private final Object synchLock = new Object();
private final String applicationRoot;
private final ILogger logger;
private MqttAsyncClient mqttClient;
private HashMap<String, ConnectionBinder> recipients = new HashMap<String, ConnectionBinder>();
private MessageStash stash;
private boolean connecting;
private String brokerUrl = null;
private String userName;
private String password;
问题的原因是ActiveMQ的MQTT连接器没有查看订阅的QoS标志。它包含以下代码
QoS qoS;
if (message.propertyExists(QOS_PROPERTY_NAME)) {
int ordinal = message.getIntProperty(QOS_PROPERTY_NAME);
qoS = QoS.values()[ordinal];
} else {
qoS = message.isPersistent() ? QoS.AT_MOST_ONCE : QoS.AT_LEAST_ONCE;
}
因此，解决问题就足以为每条消息设置QoS。如果您使用NMS：
对于ActiveMQ 5.10
msg.Properties.SetInt("ActiveMQ.MQTT.QoS", 2); // 2 - EXACTLY_ONCE
对于ActiveMQ 5.9
msg.Properties.SetInt("QoSPropertyName", 2); // 2 - EXACTLY_ONCE
使用ActiveMQ Web控制台无法实现。
花了很多时间来寻找这个问题的答案。在我看来，ActiveMq中存在一个错误。这条线
qoS = message.isPersistent() ? QoS.AT_MOST_ONCE : QoS.AT_LEAST_ONCE;
应改为
qoS = message.isPersistent() ? QoS.AT_LEAST_ONCE : QoS.AT_MOST_ONCE ;