MQTT协议之ApacheApollo服务

一、说明

MQTT是IBM开发的一个即时通讯协议，有可能成为物联网的重要组成部分。该协议支持所有平台，几乎可以把所有联网物品和外部连接起来，被用来当做传感器和致动器（比如通过Twitter让房屋联网）的通信协议。

Apache Apollo是一个代理服务器，其是在ActiveMQ基础上发展而来的，可以支持STOMP, AMQP, MQTT, Openwire, SSL, and WebSockets 等多种协议。

总结来说MQTT只是一种消息推送的协议目前(2016/1/13)为V3.1版本，而Apache Apollo是更具这种协议而开发的一款服务性的服务程序，被用来进行消息推送。同样的服务还有Mosquitto,介绍地址:http://blog.csdn.net/xukai871105/article/details/39252653

二、原理

Apache Apollo说白了其实很简单，就是在服务器端创建一个唯一订阅号，发送者可以向这个订阅号中发东西，然后接受者（即订阅了这个订阅号的人）都会收到这个订阅号发出来的消息。以此来完成消息的推送。服务器其实是一个消息中转站。

三、下载安装Apollo（windows）

1.下载地址：http://activemq.apache.org/apollo/

2.安装

apollo中间件其实是免安装的，我们只需要下载apache-apollo-1.7.1-windows-distro.zip,然后解压到某个文件夹就可以了。在这里我解压到D:\apache-apollo-1.7.1。解压开的路径如下：

3.创建属于自己的apollo域

下载下来的是官方的apollo，我们需要自己生成自己的apollo，这样做的好处就是我们可以根据自己的需求修改一些配置文件，创建过程如下：　

1 创建一个D:\myApollo文件夹。

2 进入命令输入模式，进入到刚创建的文件下下：cd D:\myApollo。

3 因为接到的目录关系所以可能有些改变，目录为（解压路径\bin\apollo create myapollo） 例子：D:\apache-apollo-1.7.1\bin\apollo create myapollo

4 创建成功后，在D:\myApollo会有一个myapollo子文件夹，里面内容如下：

其中 bin为启动目录 etc为配置目录。

4.启动myapollo

1 cd D:\myApollo\myapollo\bin （命名模式下进入到自己生成的apollo下的bin目录）

2 apollo-broker run （输入启动命令）

输入后效果如下：

这里我们可以看到端口配置信息

5.访问控制台　

在浏览器输入http://127.0.0.1:61680/，就是上面黑窗口最后一行，打开如下页面

然后输入默认用户名/密码：（admin/password），用户名密码可以在etc/users.properties中找到，点击登陆，然后进入控制页面，可以看到myapollo,当然目前里面是没有连接，没有消息的，因为我们还没有建立连接，发送消息。

至此，我们的apollo中间件就可以正常使用了。

6.修改配置

在我们的配置目录下（D:\myApollo\myapollo\etc）

apollo.xml 为网络配置信息，其他的不用管最主要的

lt;connector /> connection_limit连接限制条数2000，就是说超过2000就GG了。可不可以修改等连接到了2000条的时候更改试试。

groups.properties 用于增加用户

原本为： admins=admi

增加test用户： admins=admin|test（中间用|分开）

users.properties 用于设置用户的账号密码

用户名=密码

原本为： admin=password

增加test用户： admin=password

test=test (下面新增一列,此处与groups.properties文件对应)

7.连接程序：

依赖包：mqtt-client-0.4.0.jar

包下载地址：https://repo.eclipse.org/content/repositories/paho/org/eclipse/paho/mqtt-client/0.4.0/

代码示例下载：http://pan.baidu.com/s/1kUmsVrT （gbk编码）

在线代码解析:

public class MyMqtt{
/**连接访问者id，不能重复*/
private String clientId;
/**默认连接路径,服务器所在ip*/
private String host="tcp://192.168.1.130:61613";
/**MQTT访问默认用户名*/
private String userName = "admin";
/**MQTT访问默密码*/
private String passWord = "password";
/**发送的订阅号集合*/
private String[] publishTopics={"test"};
/**订阅者的订阅号集合*/
private String[] subTopics={"test"};
/**消息发送时的消息模式集合*/
private int[] publishQos={2};
/**消息订阅式的消息模式集合*/
private int[] subQos={2};
/**单次MQTT连接*/
private MqttClient client;
/**连接时的一些额外设置*/
private MqttConnectOptions options;
/**发送消息是的个体topic*/
private MqttTopic topic;
/**消息传递hander*/
private Handler mHandler;


public MyMqtt(Context context,final Handler handler) {
clientId=Tool.getIdentifyNumber(context);
if(client==null || !client.isConnected()){
try {
//设置连接指定的额ip，连接人
client = new MqttClient(host, clientId, new MemoryPersistence());
//开始设置连接时的参数
options = new MqttConnectOptions();
//设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接
options.setCleanSession(true);
//设置连接用户名
options.setUserName(userName);
//设置连接密码
options.setPassword(passWord.toCharArray());
//设置超时连接
options.setConnectionTimeout(10);
//设置心跳间隔
options.setKeepAliveInterval(20);
//设置当连接断开时发送的死亡预告，当这句被接受到时 证明本连接断开
//options.setWill(publishTopic, (clientId+"destroy").getBytes(), 0, true);
//连接回调函数
client.setCallback(new MqttCallback() {
@Override
public void messageArrived(String topicName, MqttMessage message) throws Exception {
// TODO Auto-generated method stub
Message msg = new Message();
msg.what=1;
msg.obj=message.toString();
handler.sendMessage(msg);
}
@Override
public void deliveryComplete(IMqttDeliveryToken token) {
// TODO Auto-generated method stub

}

@Override
public void connectionLost(Throwable cause) {
// TODO Auto-generated method stub

}
});
client.connect(options);
client.subscribe(subTopics, subQos);
} catch (MqttException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}

/**
 * 发送消息
 * @param msg
 */
public void sendMessage(String msg){
if(client!=null && client.isConnected()){
System.out.println("MQTT发送消息了呀 "+msg);
try {
MqttMessage message=new MqttMessage();
//设置消息传输的类型 0,1,2可选
message.setQos(2);
//设置是否在服务器中保存消息体
message.setRetained(false);
//设置消息的内容
message.setPayload(msg.getBytes());
//循环发送，因为一次只能一个
for (String publishTopic: publishTopics) {
topic=client.getTopic(publishTopic);
MqttDeliveryToken token = topic.publish(message);
token.waitForCompletion();
}
} catch (MqttPersistenceException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (MqttException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (NullPointerException e) {
// TODO: handle exception
e.printStackTrace();
}
}
}

8. API查看地址

http://www.eclipse.org/paho/files/javadoc/org/eclipse/paho/client/mqttv3/package-summary.html