官网：http://activemq.apache.org/apollo/index.html
环境：ubuntu 16.04
1.安装好jdk
2.下载解压
wget http://mirrors.hust.edu.cn/apache/activemq/activemq-apollo/1.7.1/apache-apollo-1.7.1-unix-distro.tar.gz
tar -zxvf apache-apollo-1.7.1-unix-distro.tar.gz
3.创建broker
(在当前用户工作目录创建即可)
apache-apollo-1.7.1/bin/apollo create apollo_broker
如果参照：http://activemq.apache.org/apollo/documentation/getting-started.html
在/var/lib目录创建broker，则创建和运行一般都需要root权限才可以；
4.修改配置，允许指定主机访问web管理控制台
参考：http://activemq.apache.org/apollo/documentation/user-manual.html#Web_Based_Administration
apollo_broker/etc/apollo.xml文件，不绑定回送地址127.0.0.1
apollo_broker/etc/login.config文件，添加白名单文件
允许以下主机访问web控制台
5.运行broker
apollo_broker/bin/apollo-broker run
IP地址：192.168.1.121
6.访问web控制台
http://192.168.1.121:61680/ (http地址，好像需要配置白名单)
或者：https://192.168.1.121:61681/ (https地址，不用配置白名单，就可以直接访问)
默认账户：admin 默认密码：password
7.创建Publish Demo
参考：apache-apollo-1.7.1/examples目录中，自带了demo
创建spring boot project
pom.xml，添加依赖组件
org.fusesource.mqtt-client
mqtt-client
1.0
org.eclipse.paho
org.eclipse.paho.client.mqttv3
1.2.0
Publish.java
package com.test;
import java.net.URISyntaxException;
import org.fusesource.hawtbuf.Buffer;
import org.fusesource.hawtbuf.UTF8Buffer;
import org.fusesource.mqtt.client.Future;
import org.fusesource.mqtt.client.FutureConnection;
import org.fusesource.mqtt.client.MQTT;
import org.fusesource.mqtt.client.QoS;
class Publisher {
public static void main(String[] args) {
System.out.println("publish main");
String user = "admin";
String password = "password";
String host = "192.168.1.121";
int port = 61613;
final String destination = "/topic/monitor";
MQTT mqtt = new MQTT();
try {
mqtt.setHost(host, port);
mqtt.setUserName(user);
mqtt.setPassword(password);
// 1.连接server
FutureConnection connection = mqtt.futureConnection();
connection.connect().await();
// 2.发布消息
UTF8Buffer topic = new UTF8Buffer(destination);
while (true) {
Buffer msg = new UTF8Buffer(getMessage());
Futurefuture = connection.publish(topic, msg, QoS.AT_LEAST_ONCE, false);
System.out.println("publish msg=" + msg + ", wait begin");
// 发送消息后，等待被处理
future.await();
// delay(1000);
System.out.println("publish wait end");
}
} catch (URISyntaxException e) {
e.printStackTrace();
} catch (Exception e) {
e.printStackTrace();
}
}
private static String getMessage() {
return String.valueOf(Math.random());
}
private static void delay(long time) {
try {
Thread.sleep(time);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
8.创建Subscriber Demo
创建spring boot project
Listener.java
package com.test;
import org.fusesource.hawtbuf.*;
import org.fusesource.mqtt.client.*;
class Listener {
public static void main(String[] args) throws Exception {
System.out.println("listener main");
String user = "admin";
String password = "password";
String host = "192.168.1.121";
int port = 61613;
final String destination = "/topic/monitor";
MQTT mqtt = new MQTT();
mqtt.setHost(host, port);
mqtt.setUserName(user);
mqtt.setPassword(password);
final CallbackConnection connection = mqtt.callbackConnection();
// 1.设置监听器
connection.listener(new org.fusesource.mqtt.client.Listener() {
long count = 0;
public void onConnected() {
System.out.println("onConnected");
}
public void onDisconnected() {
System.out.println("onDisconnected");
}
public void onFailure(Throwable value) {
System.out.println("onFailure");
value.printStackTrace();
System.exit(-2);
}
public void onPublish(UTF8Buffer topic, Buffer msg, Runnable ack) {
String message = msg.utf8().toString();
System.out.println("onPublish body=" + message + " count=" + count);
count++;
}
});
// 2.连接server
connection.connect(new Callback() {
@Override
public void onSuccess(Void value) {
System.out.println("onSuccess");
// 3.订阅消息
Topic[] topics = { new Topic(destination, QoS.AT_LEAST_ONCE) };
connection.subscribe(topics, new Callback() {
public void onSuccess(byte[] qoses) {
// UTF8Buffer buffer = new UTF8Buffer(qoses);
System.out.println("onSuccess");
}
public void onFailure(Throwable e) {
System.out.println("subscribe onFailure");
e.printStackTrace();
System.exit(-2);
}
});
}
@Override
public void onFailure(Throwable e) {
System.out.println("connect onFailure");
e.printStackTrace();
System.exit(-2);
}
});
// Wait forever..
synchronized (Listener.class) {
while (true) {
Listener.class.wait();
}
}
}
}