add rabbitmq monitor

@@ -12,5 +12,28 @@
     <artifactId>monitor-service-middleware</artifactId>
     <artifactId>monitor-service-middleware</artifactId>
     <version>1.0-SNAPSHOT</version>
     <version>1.0-SNAPSHOT</version>
 
 
+    <dependencies>
+        <!-- internal -->
+        <dependency>
+            <groupId>monitor-service</groupId>
+            <artifactId>monitor-service-common</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>monitor-service</groupId>
+            <artifactId>monitor-service-cmdb</artifactId>
+        </dependency>
+
+        <!-- 3rd -->
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.influxdb</groupId>
+            <artifactId>influxdb-java</artifactId>
+        </dependency>
+    </dependencies>
 
 
 </project>
 </project>

-package com.monitor.middleware.rabbitmq;
-
-/**
- * Created by zhengyouwei on 2016/6/20.
- */
-public class Test {
-}

+package com.monitor.middleware.rabbitmq.component;
+
+import com.monitor.middleware.rabbitmq.constant.InterVar;
+import com.monitor.middleware.rabbitmq.model.PointView;
+import org.influxdb.InfluxDB;
+import org.influxdb.InfluxDBFactory;
+import org.influxdb.dto.BatchPoints;
+import org.influxdb.dto.Point;
+import org.influxdb.dto.Query;
+import org.influxdb.dto.QueryResult;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+import java.util.List;
+
+/**
+ * Created by yoho on 2016/6/21.
+ */
+@Component
+public class InfluxComp {
+
+    public static final Logger DEBUG = LoggerFactory.getLogger(InfluxComp.class);
+
+    @Value("influxUrl")
+    private String influxUrl;
+
+    @Value("influxUser")
+    private String influxUser;
+
+    @Value("influxPwd")
+    private String influxPwd;
+
+    private InfluxDB influxDBClient;
+
+    public InfluxComp() {
+        influxDBClient = InfluxDBFactory.connect(this.influxUrl, this.influxUser, this.influxPwd);
+
+    }
+
+    @PostConstruct
+    public void doService() {
+
+        while (true) {
+
+            if (!InterVar.POINT_QUEUE.isEmpty()) {
+
+                PointView pointView = InterVar.POINT_QUEUE.poll();
+
+                influxDBClient.createDatabase(InterVar.DBNAME);
+
+                influxDBClient.write(pointView.toPoint());
+
+            }
+
+            try {
+                Thread.sleep(100L);
+
+            } catch (InterruptedException e) {
+
+                DEBUG.error("Interrupt watch point_queue , error {}", e);
+            }
+        }
+    }
+
+
+    public QueryResult doQuery(String command) {
+
+        Query query=new Query(command,InterVar.DBNAME);
+
+        return influxDBClient.query(query);
+    }
+}

+package com.monitor.middleware.rabbitmq.constant;
+
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.model.MObjectInfo;
+import com.monitor.middleware.rabbitmq.model.PointView;
+import org.influxdb.dto.BatchPoints;
+
+import javax.print.DocFlavor;
+import java.util.concurrent.*;
+
+/**
+ * Created by yoho on 2016/6/21.
+ */
+public interface InterVar {
+
+    ObjectMapper JSONMAPPER = new ObjectMapper().configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+
+    ConcurrentHashMap<Integer, MObjectInfo> moMaps = new ConcurrentHashMap<>();
+
+    ConcurrentHashMap<Integer, Integer> nodesCountMaps = new ConcurrentHashMap<>();
+
+    ConcurrentHashMap<Integer, Integer> queueCountMaps = new ConcurrentHashMap<>();
+
+    ConcurrentLinkedQueue<PointView> POINT_QUEUE = new ConcurrentLinkedQueue<>();
+
+    ExecutorService EXECUTOR_SERVICE = Executors.newFixedThreadPool(50);
+
+    String OVERVIEW_URL = "/api/overview";
+
+    String NODES_URL = "/api/nodes";
+
+    String QUEUE_URL = "/api/queues";
+
+    String MOMATCHPREFIX = "*rabbitmq*";
+
+    String URLFORMAT = "http://{}:{}";
+
+    String DBNAME = "rabbitmq_info";
+
+    String OVERVIEW_MEASURE = "overview_info";
+
+    String CLUSTERVIEW_MEASURE = "cluster_info";
+
+    String QUEUEVIEW_MEASURE = "queue_info";
+
+    String DBTAG = "moId";
+
+    int MONIT_TIMEOUT = 10;
+
+    String OVERVIEW_SQL="SELECT * FROM " + InterVar.OVERVIEW_MEASURE + " WHERE moId = \'{}\' ORDER BY time DESC LIMIT 1";
+
+    String CLUSTERVIEW_SQL="SELECT * FROM " + InterVar.CLUSTERVIEW_MEASURE +  "WHERE moId = \'{}\' ORDER BY time DESC LIMIT ";
+
+    String QUEUEVIEW_SQL="SELECT * FROM " + InterVar.QUEUEVIEW_MEASURE +  "WHERE moId = \'{}\' ORDER BY time DESC LIMIT ";
+
+}

+package com.monitor.middleware.rabbitmq.model;
+
+import com.monitor.middleware.rabbitmq.constant.InterVar;
+import lombok.Data;
+import org.influxdb.dto.BatchPoints;
+import org.influxdb.dto.Point;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Created by yoho on 2016/6/21.
+ */
+@Data
+public class ClusterView implements PointView {
+
+    private int moId;
+
+    List<NodeInfo> clusterView;
+
+    @Override
+    public BatchPoints toPoint() {
+        BatchPoints pointBp = BatchPoints.database(InterVar.DBNAME).retentionPolicy("3d").build();
+
+        for (NodeInfo nodeInfo : clusterView) {
+
+            Point point = Point.measurement(InterVar.CLUSTERVIEW_MEASURE)
+                    .tag(InterVar.DBTAG, Integer.toString(this.getMoId()))
+                    .addField("name", nodeInfo.getName())
+                    .addField("state", nodeInfo.isState() ? "running" : "idle")
+                    .addField("fd_used", nodeInfo.getFd_used())
+                    .addField("fd_total", nodeInfo.getFd_total())
+                    .addField("mem_used", nodeInfo.getMem_used())
+                    .addField("mem_limit", nodeInfo.getMem_limit())
+                    .addField("mem_alarm", nodeInfo.isMem_alarm())
+                    .addField("disk_free", nodeInfo.getDisk_free())
+                    .addField("disk_free_limit", nodeInfo.getDisk_free_limit())
+                    .addField("disk_free_alarm", nodeInfo.isDisk_free_alarm())
+                    .addField("proc_used", nodeInfo.getProc_used())
+                    .addField("proc_total", nodeInfo.getProc_total())
+                    .addField("sockets_used", nodeInfo.getSockets_used())
+                    .addField("sockets_total", nodeInfo.getSockets_total())
+                    .addField("os_pid", nodeInfo.getOs_pid())
+                    .addField("io_read_bytes", nodeInfo.getIo_read_bytes())
+                    .addField("io_write_bytes", nodeInfo.getIo_write_bytes())
+                    .addField("io_read_bytes_rate", nodeInfo.getIo_read_bytes_rate().getRate())
+                    .addField("io_write_bytes_rate", nodeInfo.getIo_write_bytes_rate().getRate())
+                    .addField("ip", nodeInfo.getClusters().get(0).getIp())
+                    .addField("port", nodeInfo.getClusters().get(0).getPort())
+                    .build();
+
+            pointBp.point(point);
+
+        }
+        InterVar.nodesCountMaps.put(moId, clusterView.size());
+        return pointBp;
+    }
+}

+package com.monitor.middleware.rabbitmq.model;
+
+import lombok.Data;
+
+/**
+ * Created by yoho on 2016/6/21.
+ */
+@Data
+public class GlobalCount {
+    int connections;
+
+    int channels;
+
+    int exchanges;
+
+    int queues;
+
+    int consumers;
+}

+package com.monitor.middleware.rabbitmq.model;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import lombok.Data;
+
+/**
+ * Created by yoho on 2016/6/21.
+ */
+@Data
+public class Host {
+        @JsonProperty("peer_addr")
+        String ip;
+
+        @JsonProperty("peer_port")
+        int port;
+}

+package com.monitor.middleware.rabbitmq.model;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import lombok.Data;
+
+/**
+ * Created by yoho on 2016/6/21.
+ */
+@Data
+public class Message {
+
+    @JsonProperty("messages_ready")
+    int readyMsg;
+
+    @JsonProperty("messages_unacknowledged")
+    int unackMsg;
+
+    @JsonProperty("messages")
+    int totalMsg;
+}

+package com.monitor.middleware.rabbitmq.model;
+
+import lombok.Data;
+
+/**
+ * Created by yoho on 2016/6/21.
+ */
+@Data
+public class MoInfo {
+
+    private int moId;
+
+    private String moName;
+
+    private int moTypes;
+
+    private int mo_ip;
+}

+package com.monitor.middleware.rabbitmq.model;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.monitor.middleware.rabbitmq.constant.InterVar;
+import lombok.Data;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static com.monitor.middleware.rabbitmq.constant.InterVar.JSONMAPPER;
+
+/**
+ * Created by yoho on 2016/6/21.
+ */
+@Data
+public class NodeInfo {
+    String name;
+
+    @JsonProperty("running")
+    boolean state;
+
+    int fd_used;
+
+    int fd_total;
+
+    long mem_used;
+
+    long mem_limit;
+
+    boolean mem_alarm;
+
+    long disk_free;
+
+    long disk_free_limit;
+
+    boolean disk_free_alarm;
+
+    int proc_used;
+
+    int proc_total;
+
+    int sockets_used;
+
+    int sockets_total;
+
+    int os_pid;
+
+    long io_read_bytes;
+
+    long io_write_bytes;
+
+  //need set un-auto
+    @JsonProperty("io_read_bytes_details")
+    IoReadRate io_read_bytes_rate;
+
+    @JsonProperty("io_write_bytes_details")
+    IoWriteRate io_write_bytes_rate;
+
+    @JsonProperty("cluster_links")
+       List<Host> clusters;
+
+    @Data
+    class IoReadRate{
+        double rate;
+    }
+
+    @Data
+    class IoWriteRate{
+        double rate;
+    }
+
+}
+
+

+package com.monitor.middleware.rabbitmq.model;
+
+import com.monitor.middleware.rabbitmq.constant.InterVar;
+import lombok.Data;
+import org.influxdb.dto.BatchPoints;
+import org.influxdb.dto.Point;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Created by yoho on 2016/6/21.
+ */
+@Data
+public class OverView implements PointView {
+
+    int moId;
+
+    GlobalCount object_totals;
+
+    Message queue_totals;
+
+    @Override
+    public BatchPoints toPoint() {
+
+        BatchPoints pointBp = BatchPoints.database(InterVar.DBNAME).retentionPolicy("3d").build();
+
+        Point point = Point.measurement(InterVar.OVERVIEW_MEASURE)
+                .tag(InterVar.DBTAG, Integer.toString(this.getMoId()))
+                .addField("connections", this.getObject_totals().getConnections())
+                .addField("channels", this.getObject_totals().getChannels())
+                .addField("exchanges", this.getObject_totals().getExchanges())
+                .addField("queues", this.getObject_totals().getQueues())
+                .addField("consumers", this.getObject_totals().getConsumers())
+                .addField("messages_ready", this.getQueue_totals().getReadyMsg())
+                .addField("messages_unacknowledged", this.getQueue_totals().getUnackMsg())
+                .addField("messages", this.getQueue_totals().getTotalMsg())
+                .build();
+
+        pointBp.point(point);
+
+        return pointBp;
+    }
+}
+

+package com.monitor.middleware.rabbitmq.model;
+
+import org.influxdb.dto.BatchPoints;
+
+/**
+ * Created by yoho on 2016/6/21.
+ */
+public interface PointView {
+
+    BatchPoints toPoint();
+}

+package com.monitor.middleware.rabbitmq.model;
+
+import lombok.Data;
+
+/**
+ * Created by yoho on 2016/6/21.
+ */
+@Data
+public class QueryRequest {
+
+    String moId;
+}

+package com.monitor.middleware.rabbitmq.model;
+
+import lombok.Data;
+
+/**
+ * Created by yoho on 2016/6/21.
+ */
+@Data
+public class QueueInfo {
+
+    String name;
+
+    String vhost;
+
+    String node;
+
+    int messages;
+
+    int messages_ready;
+
+    int messages_unacknowledged;
+
+    int consumers;
+
+    String state;
+
+    String idle_since;
+
+    boolean isWatch = false;
+}

+package com.monitor.middleware.rabbitmq.model;
+
+import com.monitor.middleware.rabbitmq.constant.InterVar;
+import lombok.Data;
+import org.influxdb.dto.BatchPoints;
+import org.influxdb.dto.Point;
+
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Created by yoho on 2016/6/21.
+ */
+@Data
+public class QueueView implements PointView {
+    int moId;
+
+    List<QueueInfo> queueView;
+
+    @Override
+    public BatchPoints toPoint() {
+
+        BatchPoints pointBp = BatchPoints.database(InterVar.DBNAME).retentionPolicy("3d").build();
+
+        for (QueueInfo queueInfo : queueView) {
+
+            Point point = Point.measurement(InterVar.QUEUEVIEW_MEASURE)
+                    .tag(InterVar.DBTAG, Integer.toString(this.getMoId()))
+                    .addField("name", queueInfo.getName())
+                    .addField("vhost", queueInfo.getVhost())
+                    .addField("node", queueInfo.getNode())
+                    .addField("messages", queueInfo.getMessages())
+                    .addField("messages_ready", queueInfo.getMessages_ready())
+                    .addField("messages_unacknowledged", queueInfo.getMessages_unacknowledged())
+                    .addField("messages_ready", queueInfo.getMessages_ready())
+                    .addField("state", queueInfo.getState())
+                    .addField("idle_since", queueInfo.getIdle_since())
+                    .build();
+
+            pointBp.point(point);
+        }
+
+        InterVar.queueCountMaps.put(moId, queueView.size());
+
+        return pointBp;
+    }
+}

+package com.monitor.middleware.rabbitmq.service;
+
+import com.monitor.middleware.rabbitmq.component.InfluxComp;
+import com.monitor.middleware.rabbitmq.constant.InterVar;
+import com.monitor.middleware.rabbitmq.model.QueryRequest;
+import com.monitor.model.response.BaseResponse;
+import org.influxdb.InfluxDB;
+import org.influxdb.dto.Query;
+import org.influxdb.dto.QueryResult;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+/**
+ * Created by yoho on 2016/6/21.
+ */
+@RestController(value = "/middleware/rabbitmq")
+public class RabbitmqService {
+    public static final Logger DEBUG = LoggerFactory.getLogger(RabbitmqService.class);
+
+    @Autowired
+    InfluxComp influxComp;
+
+    @RequestMapping(value = "/overview")
+    public BaseResponse queryOverview(@RequestBody QueryRequest request) {
+
+        BaseResponse response = new BaseResponse();
+        QueryResult result = null;
+        try {
+            result = this.influxComp.doQuery(String.format(InterVar.OVERVIEW_SQL, request.getMoId()));
+        } catch (Exception e) {
+            DEBUG.error("Failed to query overview about mo {}", request.getMoId());
+            response.setData(e);
+        }
+        response.setData(result);
+
+        return response;
+    }
+
+
+    @RequestMapping(value = "/cluster")
+    public BaseResponse queryCluster(@RequestBody QueryRequest request) {
+
+        BaseResponse response = new BaseResponse();
+        QueryResult result = null;
+        try {
+            result = this.influxComp.doQuery(String.format(InterVar.CLUSTERVIEW_SQL, request.getMoId()) + InterVar.nodesCountMaps.get(request.getMoId()));
+        } catch (Exception e) {
+            DEBUG.error("Failed to query overview about mo {}", request.getMoId());
+            response.setData(e);
+        }
+        response.setData(result);
+
+        return response;
+    }
+
+
+    @RequestMapping(value = "/queue")
+    public BaseResponse queryQueue(@RequestBody QueryRequest request) {
+
+        BaseResponse response = new BaseResponse();
+        QueryResult result = null;
+        try {
+            result = this.influxComp.doQuery(String.format(InterVar.QUEUEVIEW_SQL, request.getMoId()) + InterVar.queueCountMaps.get(request.getMoId()));
+
+
+        } catch (Exception e) {
+            DEBUG.error("Failed to query overview about mo {}", request.getMoId());
+            response.setData(e);
+        }
+        response.setData(result);
+
+        return response;
+    }
+
+}

+package com.monitor.middleware.rabbitmq.task;
+
+import com.model.MObjectInfo;
+import com.model.TypeInfo;
+import com.monitor.cmdb.service.IMObjectInfoService;
+import com.monitor.cmdb.service.ITypeInfoService;
+import com.monitor.middleware.rabbitmq.constant.InterVar;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.scheduling.annotation.Scheduled;
+
+import java.util.List;
+
+/**
+ * Created by yoho on 2016/6/21.
+ */
+public class MoScanTask  implements Runnable{
+
+    @Autowired
+    IMObjectInfoService moService;
+
+    @Autowired
+    ITypeInfoService typeService;
+
+    //自动发现rabbit监控对象
+    @Scheduled(cron = "")
+    public void doTask() {
+        List<MObjectInfo> mObjectInfoList = moService.queryMObjectsInfo();
+
+        for (MObjectInfo info : mObjectInfoList) {
+
+            TypeInfo typeInfo = typeService.queryTypeInfo(info.getMoTypeId());
+
+            if (1 == typeInfo.getTypeIsLeaf() && typeInfo.getTypeName().matches(InterVar.MOMATCHPREFIX)) {
+
+                InterVar.moMaps.put(info.getMoId(), info);
+
+            }
+        }
+    }
+
+    @Override
+    public void run() {
+        doTask();
+    }
+}

+package com.monitor.middleware.rabbitmq.task;
+
+import com.model.MObjectInfo;
+import com.monitor.middleware.rabbitmq.constant.InterVar;
+import com.monitor.middleware.rabbitmq.task.job.ClusterViewJob;
+import com.monitor.middleware.rabbitmq.task.job.OneJob;
+import com.monitor.middleware.rabbitmq.task.job.OverViewJob;
+import com.monitor.middleware.rabbitmq.task.job.QueueViewJob;
+import org.springframework.scheduling.annotation.Scheduled;
+
+import java.util.Map;
+
+import static com.monitor.middleware.rabbitmq.constant.InterVar.EXECUTOR_SERVICE;
+
+/**
+ * Created by yoho on 2016/6/21.
+ */
+
+/**
+ * 加载服务监控任务
+ */
+public class MonitTask {
+
+    @Scheduled(cron = "")
+    public void doTask() {
+        for (Map.Entry<Integer, MObjectInfo> entry : InterVar.moMaps.entrySet()) {
+            EXECUTOR_SERVICE.submit(new OneJob(new ClusterViewJob(entry.getKey())));
+            EXECUTOR_SERVICE.submit(new OneJob(new OverViewJob(entry.getKey())));
+            EXECUTOR_SERVICE.submit(new OneJob(new QueueViewJob(entry.getKey())));
+        }
+    }
+}

+package com.monitor.middleware.rabbitmq.task.job;
+
+
+import com.model.MObjectInfo;
+import com.monitor.middleware.rabbitmq.constant.InterVar;
+import com.monitor.middleware.rabbitmq.model.ClusterView;
+import com.monitor.middleware.rabbitmq.model.NodeInfo;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.client.RestTemplate;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.concurrent.Callable;
+
+/**
+ * Created by yoho on 2016/6/21.
+ */
+
+public class ClusterViewJob implements Callable {
+    private final static Logger DEBUG = LoggerFactory.getLogger(ClusterViewJob.class);
+
+    @Autowired
+    RestTemplate restClient;
+
+    private int moId;
+
+    public ClusterViewJob(int moId) {
+
+        this.moId = moId;
+    }
+
+    public void doTask() {
+        MObjectInfo moInfo = InterVar.moMaps.get(moId);
+
+        String url = String.format(InterVar.URLFORMAT, moInfo.getMoHostIp(), moInfo.getMoTags()) + InterVar.NODES_URL;
+
+        if (null == moInfo) {
+
+            return;
+        }
+
+        String respJson = restClient.getForObject(url, String.class);
+
+        ClusterView oneView = new ClusterView();
+
+        try {
+            oneView.setMoId(moId);
+
+            oneView.setClusterView(Arrays.asList(InterVar.JSONMAPPER.readValue(respJson, NodeInfo[].class)));
+
+
+        } catch (IOException e) {
+
+            DEBUG.error("Failed to parse cluster_view info: {} , error {}", respJson, e);
+
+            return;
+        }
+
+        if (null != oneView) {
+
+            InterVar.POINT_QUEUE.offer(oneView);
+        }
+    }
+
+    @Override
+    public Object call() throws Exception {
+        doTask();
+        return null;
+    }
+}

+package com.monitor.middleware.rabbitmq.task.job;
+
+import com.monitor.middleware.rabbitmq.constant.InterVar;
+import java.util.concurrent.*;
+
+
+
+/**
+ * Created by yoho on 2016/6/21.
+ */
+public class OneJob implements Runnable {
+
+    private Callable oneJob;
+
+    public  OneJob(Callable oneJob) {
+        this.oneJob = oneJob;
+    }
+
+
+    @Override
+    public void run() {
+        Future future = InterVar.EXECUTOR_SERVICE.submit(this.oneJob);
+
+        try {
+            future.get(InterVar.MONIT_TIMEOUT, TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        } catch (ExecutionException e) {
+            e.printStackTrace();
+        } catch (TimeoutException e) {
+            e.printStackTrace();
+        }
+    }
+}

+package com.monitor.middleware.rabbitmq.task.job;
+
+
+import com.model.MObjectInfo;
+import com.monitor.middleware.rabbitmq.constant.InterVar;
+import com.monitor.middleware.rabbitmq.model.OverView;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.client.RestTemplate;
+
+import java.io.IOException;
+import java.util.concurrent.Callable;
+
+/**
+ * Created by yoho on 2016/6/21.
+ */
+
+public class OverViewJob implements Callable{
+    private final static Logger DEBUG = LoggerFactory.getLogger("OverViewJob");
+
+    @Autowired
+    RestTemplate restClient;
+
+    private int moId;
+
+    public OverViewJob(int moId) {
+
+        this.moId = moId;
+    }
+
+    public void doTask() {
+        MObjectInfo moInfo = InterVar.moMaps.get(moId);
+
+        String url = String.format(InterVar.URLFORMAT, moInfo.getMoHostIp(), moInfo.getMoTags()) + InterVar.OVERVIEW_URL;
+
+        if (null == moInfo) {
+
+            return;
+        }
+
+        String respJson = restClient.getForObject(url, String.class);
+
+        OverView oneView = null;
+
+        try {
+            oneView = InterVar.JSONMAPPER.readValue(respJson, OverView.class);
+
+            oneView.setMoId(moInfo.getMoId());
+
+        } catch (IOException e) {
+
+            DEBUG.error("Failed to parse overview info: {} , error {}", respJson, e);
+
+            return;
+        }
+
+        if (null != oneView) {
+            InterVar.POINT_QUEUE.offer(oneView);
+        }
+    }
+
+    @Override
+    public Object call() throws Exception {
+        doTask();
+        return null;
+    }
+}

+package com.monitor.middleware.rabbitmq.task.job;
+
+
+import com.model.MObjectInfo;
+import com.monitor.middleware.rabbitmq.constant.InterVar;
+import com.monitor.middleware.rabbitmq.model.QueueInfo;
+import com.monitor.middleware.rabbitmq.model.QueueView;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.client.RestTemplate;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.concurrent.Callable;
+
+/**
+ * Created by yoho on 2016/6/21.
+ */
+
+public class QueueViewJob implements Callable {
+    private final static Logger DEBUG = LoggerFactory.getLogger(QueueViewJob.class);
+
+    @Autowired
+    RestTemplate restClient;
+
+    private int moId;
+
+    public QueueViewJob(int moId) {
+
+        this.moId = moId;
+    }
+
+    public void doTask() {
+        MObjectInfo moInfo = InterVar.moMaps.get(moId);
+
+        String url = String.format(InterVar.URLFORMAT, moInfo.getMoHostIp(), moInfo.getMoTags()) + InterVar.QUEUE_URL;
+
+        if (null == moInfo) {
+
+            return;
+        }
+
+        String respJson = restClient.getForObject(url, String.class);
+
+        QueueView oneView = new QueueView();
+
+        try {
+            oneView.setQueueView(Arrays.asList(InterVar.JSONMAPPER.readValue(respJson, QueueInfo[].class)));
+
+            oneView.setMoId(moInfo.getMoId());
+
+        } catch (IOException e) {
+
+            DEBUG.error("Failed to parse queue_view info: {} , error {}", respJson, e);
+
+            return;
+        }
+
+        if (null != oneView) {
+
+            InterVar.POINT_QUEUE.offer(oneView);
+        }
+    }
+
+
+    @Override
+    public Object call() throws Exception {
+        doTask();
+        return null;
+    }
+}

+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+       xmlns:task="http://www.springframework.org/schema/task"
+       xsi:schemaLocation="http://www.springframework.org/schema/beans
+       http://www.springframework.org/schema/beans/spring-beans-3.1.xsd
+       http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd">
+
+    <task:scheduler id="taskScheduler" pool-size="5"></task:scheduler>
+
+    <bean id="httpClientFactory" class="org.springframework.http.client.SimpleClientHttpRequestFactory">
+        <property name="connectTimeout" value="2000"/>
+        <property name="readTimeout" value="3000"/>
+    </bean>
+
+    <!--RestTemplate-->
+    <bean id="restClient" class="org.springframework.web.client.RestTemplate" scope="prototype">
+        <constructor-arg ref="httpClientFactory"/>
+    </bean>
+
+
+
+</beans>