Question

Apache apollo 怎么实现集群部署

Answer 1

1、下载Apollo服务器，下载后解压，前桥然后运行apache-apollo-1.6\bin\apollo.cmd，输入createmybroker（名字任意取，这里是根据官网介绍的来取的）创建服务器实例，服务器实例包含了所有的配置，运行时数据等，并且和一个服务器进程关联。　　2、createmybroker之后会在bin目录下生成mybroker文件夹，慧弊猛里面包含有很多信息，其中etc\apollo.xml文件下是配置服务器信息的文件，etc\users.properties文件包含连接MQTT服务器时用到的用户名和密码，后面会介绍，可以修改原始的admin=password，可以接着换行添加新的用户名密码。　　3、打开cmd，运行…apache-apollo-1.6\bin\mybroker\bin\apollo-broker.cmdrun开启服务器，可以在浏览器中输入查看是否安装成功，该界面展示了topic，连接数等很多信息。　　经过上面的简单步骤，服务器基本上就已经完成，下一篇将介绍Android客户端的编写和注意事项。　　客户端使用的API，开始我使用的是mqtt-client，使用过后发现问题百出，不能很好的满足要求，后来卜磨使用了官方推荐的EclipsePaho，下面开始客户端代码的编写，为了方便测试这里有android和j2se两个工程：　　1、新建android工程MQTTClient　　2、MainActivity代码如下：　　[java]viewplaincopyprint?packageldw.mqttclient;importjava.util.concurrent.Executors;importjava.util.concurrent.ScheduledExecutorService;importjava.util.concurrent.TimeUnit;importorg.eclipse.paho.client.mqttv3.IMqttDeliveryToken;importorg.eclipse.paho.client.mqttv3.MqttCallback;importorg.eclipse.paho.client.mqttv3.MqttClient;importorg.eclipse.paho.client.mqttv3.MqttConnectOptions;importorg.eclipse.paho.client.mqttv3.MqttException;importorg.eclipse.paho.client.mqttv3.MqttMessage;importorg.eclipse.paho.client.mqttv3.persist.MemoryPersistence;importandroid.app.Activity;importandroid.os.Bundle;importandroid.os.Handler;importandroid.os.Message;importandroid.view.KeyEvent;importandroid.widget.TextView;importandroid.widget.Toast;publicclassMainActivityextendsActivity{privateTextViewresultTv;privateStringhost="tcp://127.0.0.1:1883";privateStringuserName="admin";privateStringpassWord="password";privateHandlerhandler;privateMqttClientclient;privateStringmyTopic="test/topic";privateMqttConnectOptionsoptions;privateScheduledExecutorServicescheduler;@OverrideprotectedvoidonCreate(BundlesavedInstanceState){super.onCreate(savedInstanceState);setContentView(R.layout.main);resultTv=(TextView)findViewById(R.id.result);init();handler=newHandler(){@OverridepublicvoidhandleMessage(Messagemsg){super.handleMessage(msg);if(msg.what==1){Toast.makeText(MainActivity.this,(String)msg.obj,Toast.LENGTH_SHORT).show();System.out.println("-----------------------------");}elseif(msg.what==2){Toast.makeText(MainActivity.this,"连接成功",Toast.LENGTH_SHORT).show();try{client.subscribe(myTopic,1);}catch(Exceptione){e.printStackTrace();}}elseif(msg.what==3){Toast.makeText(MainActivity.this,"连接失败，系统正在重连",Toast.LENGTH_SHORT).show();}}};startReconnect();}privatevoidstartReconnect(){scheduler=Executors.newSingleThreadScheduledExecutor();scheduler.scheduleAtFixedRate(newRunnable(){@Overridepublicvoidrun(){if(!client.isConnected()){connect();}}},0*1000,10*1000,TimeUnit.MILLISECONDS);}privatevoidinit(){try{//host为主机名，test为clientid即连接MQTT的客户端ID，一般以客户端唯一标识符表示，MemoryPersistence设置clientid的保存形式，默认为以内存保存client=newMqttClient(host,"test",newMemoryPersistence());//MQTT的连接设置options=newMqttConnectOptions();//设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接options.setCleanSession(true);//设置连接的用户名options.setUserName(userName);//设置连接的密码options.setPassword(passWord.toCharArray());//设置超时时间单位为秒options.setConnectionTimeout(10);//设置会话心跳时间单位为秒服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制options.setKeepAliveInterval(20);//设置回调client.setCallback(newMqttCallback(){@OverridepublicvoidconnectionLost(Throwablecause){//连接丢失后，一般在这里面进行重连System.out.println("connectionLost----------");}@OverridepublicvoiddeliveryComplete(IMqttDeliveryTokentoken){//publish后会执行到这里System.out.println("deliveryComplete---------"+token.isComplete());}@OverridepublicvoidmessageArrived(StringtopicName,MqttMessagemessage)throwsException{//subscribe后得到的消息会执行到这里面System.out.println("messageArrived----------");Messagemsg=newMessage();msg.what=1;msg.obj=topicName+"---"+message.toString();handler.sendMessage(msg);}});//connect();}catch(Exceptione){e.printStackTrace();}}privatevoidconnect(){newThread(newRunnable(){@Overridepublicvoidrun(){try{client.connect(options);Messagemsg=newMessage();msg.what=2;handler.sendMessage(msg);}catch(Exceptione){e.printStackTrace();Messagemsg=newMessage();msg.what=3;handler.sendMessage(msg);}}}).start();}@OverridepublicbooleanonKeyDown(intkeyCode,KeyEventevent){if(client!=null&&keyCode==KeyEvent.KEYCODE_BACK){try{client.disconnect();}catch(Exceptione){e.printStackTrace();}}returnsuper.onKeyDown(keyCode,event);}@OverrideprotectedvoidonDestroy(){super.onDestroy();try{scheduler.shutdown();client.disconnect();}catch(MqttExceptione){e.printStackTrace();}}}　　由于项目需要，我用到了心跳重连。根据这里的解释设置apollo.xml，主要有设置主机连接的地址。另外，options还有个setWill方法，如果项目中需要知道客户端是否掉线可以调用该方法。