java后端IM消息推送服务开发——协议

public class MqttProtocol 
{
	private static Logger logger = Logger.getLogger(MqttProtocol.class);  
    public static final String HOST = "tcp://xx.xx.xx.xx:1883";
    private static final String CLIENTID = "yyyy";
    private MqttClient client;
    private MqttConnectOptions options = new MqttConnectOptions();
    //private String userName = "admin";
    //private String passWord = "public";
    public MqttMessage message;
    private PushCallback callback;
    /**
     * 用于初始化mqttclient客户端，设置回调函数，同时连接mqtt服务器
     * @throws MqttException
     */
       public MqttProtocol() throws MqttException 
       {
           //MemoryPersistence设置clientid的保存形式，默认为以内存保存
           client = new MqttClient(HOST, CLIENTID, new MemoryPersistence());
           callback = new PushCallback();
           client.setCallback(callback);
           options = new MqttConnectOptions();
           options.setCleanSession(false);
           options.setKeepAliveInterval(60);
           connect();
       }
       /**
        * 连接mqtt消息服务器，同时设置了断开重连的功能，主要是为了高可用性考虑，在断网服务器崩溃时候我们的程序仍然不会终止
        */
       private void connect() 
       {
      	 SimpleDateFormat sdf= new SimpleDateFormat(Constant.DATE_FORMAT_MDYHMS);
      	 System.out.println(sdf.format(System.currentTimeMillis()));
           boolean tryConnecting = true;
           while (tryConnecting) {
             try {
               client.connect(options);
             } catch (Exception e1) {
          	   System.out.println("Connection attempt failed with '"+e1.getCause()+
                    "'. Retrying.");
             }
             if (client.isConnected()) {
          	   System.out.println("Connected.");
               tryConnecting = false;
             } else {
               pause();
             }
           }
       }
       private void pause() {
   	    try {
   	      Thread.sleep(1000);
   	    } catch (InterruptedException e) {
   	      // Error handling goes here...
   	    }
   	  }
       /**
        * 
        * @param topic
        * @param qos
        * @throws MqttPersistenceException
        * @throws MqttException
        * 订阅相关主题
        */
       public void subscribe(String topic , int qos) throws MqttPersistenceException,
       		MqttException 
       {
      	 client.subscribe(topic, qos);    	
       }
       /**
        * 
        * @throws MqttPersistenceException
        * @throws MqttException
        * 断开连接服务器
        */
       public void disconnect() throws MqttPersistenceException,
  		MqttException 
  	 {
      	 client.disconnect();
       }
       /**
        * 
        * @author binshi
        *实现mqttcallback接口，主要用于接收消息后的处理方法
        */
       private class PushCallback implements MqttCallback { 
      	 /**
      	  * 断开后 系统会自动调用这个函数，同时在这个函数里进行重连操作
      	  */
   	    public void connectionLost(Throwable cause) {  
   	        // 连接丢失后，一般在这里面进行重连  
   	    	System.out.println("连接断开，可以做重连");  
   	        connect();
   	        try {
				subscribe(Constant.TOPIC_MQTT_PROTOCOL, 2);
			} catch (MqttPersistenceException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (MqttException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
   	    } 
   	    /**
   	     * 消息成功传送后，系统会自动调用此函数，表明成功向topic发送消息
   	     */
   		@Override
  		public void deliveryComplete(IMqttDeliveryToken arg0) {
  			// TODO Auto-generated method stub
   			System.out.println("deliveryComplete---------" + arg0.isComplete());
  		}
   		/**
   		 * 连接mongo数据库，返回关于具体collection的Mongocollection
   		 * @param collectionname
   		 * @return
   		 */
   		
   		public void messageArrived(String topic, MqttMessage message) throws Exception
   		{
			System.out.println(topic);
	    	SimpleDateFormat sdf= new SimpleDateFormat(Constant.DATE_FORMAT_MDYHMS);
	    	System.out.println(sdf.format(System.currentTimeMillis()));
	    	System.out.println("接收消息主题 : " + topic);  
	    	System.out.println("接收消息Qos : " + message.getQos());  
	    	System.out.println("接收消息内容 : " + new String(message.getPayload()));
	    	//1 抽取事件信令消息
		    String messagejudge=new String(message.getPayload());
		    System.out.println("忽略所有robot消息以及offline离线消息");
		    JSONObject jo=new JSONObject();
			try {
				 jo=JSONObject.fromObject(messagejudge);		
			} catch (Exception e) {							
				e.printStackTrace();
			}
			String from=jo.getString("from");
			System.out.println("获得from"+from);
			System.out.println("确定消息是否包含offline，如果包含取得offline，为1就不处理");
			String offline=null;
			if(messagejudge.contains("offline"))
			{
				offline=jo.getString("offline");
			}
			if((offline==null)&&(!from.contains("robot")))
			{
				System.out.println("处理非系统消息和非离线消息");
				String type=jo.getString("type");
				System.out.println("获得type"+type);
				if(type.equals("shakehand"))
				{
					System.out.println("处理shakehand消息");
					String admin="doyounkowwhy";
					if(jo.toString().contains("admin"))
					{
						admin=jo.getString("admin");
					}
					System.out.println("取得admin 如果为1定义为客服，否则为普通用户 admin为"+admin);
					if(admin.equals("1"))
					{
						System.out.println("处理客服握手消息");
						System.out.println("发送握手成功消息");
						MqttTopic retopic=client.getTopic(topic);
						MsgOperation.sendSysMsgToClient(from,"0", "1005", "握手成功", null,retopic);
						System.out.println("向客户端发送离线未接收的消息");
						String convid=jo.getString("convid");
						String database="dolina";
						String collection="messages";
						MongoDBDao.getMongoDBDaoInstance().sendOfflineMsgToClient(from, convid,retopic,database,collection);
					}
					else
					{
						System.out.println("处理普通用户的握手消息");
						String appid=jo.getString("appid");
						String pageid=jo.getString("pageid");
						String convid=jo.getString("convid");
						MqttTopic retopic=client.getTopic(topic);
						MsgOperation.sendShakeHandInfo(from,convid,appid,pageid,retopic);
					}
				}
				else if(type.equals("text")||type.equals("image"))
				{
					System.out.println("处理图片和文字消息");
					String tmpindex=jo.getString("tmpindex");
					String convid=jo.getString("convid");
					MqttTopic retopic=client.getTopic(topic);
					MsgOperation.getTextMsg( tmpindex, from, convid, retopic);
					System.out.println("保存图片文字消息");
					String database="dolina";
					String collection="messages";
					MongoDBDao.getMongoDBDaoInstance().saveTextMsg(database,collection,jo);
				}
				else if(type.equals("ack"))
				{
					System.out.println("处理ack消息");
					String tmpindex=jo.getString("tmpindex");
					String convid=jo.getString("convid");
					String database="dolina";
					String collection="messages";
					MongoDBDao.getMongoDBDaoInstance().getAck(tmpindex,convid,from,database,collection);
				}
			}
		    
		   
   		}
   	}
       /**
        * 
        * @param args
        * @throws MqttException
        * 整个工程从这里开始执行，生成可执行jar包，这个设置为主类。
        */
       public static void main(String[] args) throws MqttException 
       {  
    	   MqttProtocol signal = new MqttProtocol();
           signal.message = new MqttMessage();
           /**
           server.message.setQos(2);
           server.message.setRetained(false);
           server.message.setPayload("给客户端124推送的信息".getBytes()); 
           server.subscribe("/engyne/1/7/169573fcbc96a816281192222", 2);
           */
           signal.subscribe(Constant.TOPIC_MQTT_PROTOCOL, 2);
           System.out.println(signal.message.isRetained() + "------ratained状态");
  	}
}

public class RemoteOperation
{
	private static Logger logger = Logger.getLogger(MqttProtocol.class);  
	public static JSONObject remoteCall(String url) throws HttpException, IOException
		{
			HttpClient httpClient = new HttpClient(); 
	    	GetMethod method =null ;
	    	method=new GetMethod(url);
	    	int retcode = httpClient.executeMethod(method);
	    	if (retcode != HttpStatus.SC_OK)
	    	{// 发送不成功  
	          logger.info("远程调用出错"); 
	          return null;
	        }
	    	else 
	        {  
	        	String body = method.getResponseBodyAsString();  
	        	logger.info(body+"远程调用php成功");
	        	JSONObject jsonObject=new JSONObject();
				try {
					 jsonObject=JSONObject.fromObject(body);		
				} catch (Exception e) {							
					e.printStackTrace();
				}
				if (method != null) 
			    {  
			        method.releaseConnection();  
			    } 
				return jsonObject;
	       } 
	      
	       
		}
}