新增webSocket推送服务

pom.xml

@@ -134,6 +134,23 @@
             <artifactId>junrar</artifactId>
             <version>0.7</version>
         </dependency>
+<!--        引用webSocket依赖-->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-websocket</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-lang3</artifactId>
+            <version>RELEASE</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-configuration-processor</artifactId>
+            <optional>true</optional>
+        </dependency>
+
     </dependencies>
 
     <build>

src/main/java/com/zjty/inspect/config/WebSocketConfig.java

+package com.zjty.inspect.config;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.web.socket.server.standard.ServerEndpointExporter;
+
+/**
+ * @program: liujie-cloud
+ * @description: 开启WebSocket支持
+ * @author: LiuJie
+ * @create: 2019-05-17 10:09
+ **/
+@Configuration
+public class WebSocketConfig {
+
+    @Bean
+    public ServerEndpointExporter serverEndpointExporter() {
+        return new ServerEndpointExporter();
+    }
+}
\ No newline at end of file

src/main/java/com/zjty/inspect/config/WebSocketServer.java

+package com.zjty.inspect.config;
+
+import com.alibaba.fastjson.JSON;
+import com.zjty.inspect.utils.ExceptionMessage;
+import lombok.Data;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+
+import javax.websocket.*;
+import javax.websocket.server.PathParam;
+import javax.websocket.server.ServerEndpoint;
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArraySet;
+
+/**
+ * @program: liujie-cloud
+ * @description: WebSocket服务类
+ * @author: LiuJie
+ * @create: 2019-05-17 10:08
+ **/
+
+@ServerEndpoint("/websocket/{sid}")
+@Component
+@Data
+public class WebSocketServer {
+    static Logger log= LoggerFactory.getLogger(WebSocketServer.class);
+
+    /**
+     * 静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
+     */
+    private static int onlineCount = 0;
+    /**
+     * concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
+     */
+    public static CopyOnWriteArraySet<WebSocketServer> webSocketSet = new CopyOnWriteArraySet<WebSocketServer>();
+
+    /**
+     * 与某个客户端的连接会话，需要通过它来给客户端发送数据
+     */
+    private Session session;
+
+    /**
+     * 接收sid
+     */
+    private String sid="";
+
+    /**
+     * 连接建立成功调用的方法*/
+    @OnOpen
+    public void onOpen(Session session, @PathParam("sid") String sid) {
+        this.session = session;
+        //加入set中
+        webSocketSet.add(this);
+        //在线数加1
+        addOnlineCount();
+        log.info("有新窗口开始监听:"+sid+",当前在线人数为" + getOnlineCount());
+        this.sid=sid;
+        try {
+            sendMessage("连接成功");
+        } catch (IOException e) {
+            String info = ExceptionMessage.getStackTraceInfo(e);
+            log.error("websocket IO异常:"+info);
+        }
+    }
+
+    /**
+     * 连接关闭调用的方法
+     */
+    @OnClose
+    public void onClose() {
+        //从set中删除
+        webSocketSet.remove(this);
+        //在线数减1
+        subOnlineCount();
+        log.info("有一连接关闭！当前在线人数为" + getOnlineCount());
+    }
+
+    /**
+     * 收到客户端消息后调用的方法
+     *
+     * @param message 客户端发送过来的消息*/
+    @OnMessage
+    public void onMessage(String message, Session session) {
+        log.info("收到来自窗口"+sid+"的信息:"+message);
+        /**
+         * 群发消息
+         */
+        for (WebSocketServer item : webSocketSet) {
+            try {
+                item.sendMessage(message);
+            } catch (Exception e) {
+                String info = ExceptionMessage.getStackTraceInfo(e);
+                log.error("发生错误："+info);
+            }
+        }
+    }
+
+    /**
+     *
+     * @param session
+     * @param error
+     */
+    @OnError
+    public void onError(Session session, Throwable error) {
+        log.error("发生错误");
+        String error1 = ExceptionMessage.getError(error);
+        log.info("error"+error1);
+        error.printStackTrace();
+    }
+   /**
+   * @Description: 实现服务器主动推送
+   * @Param:
+   * @return:
+   * @Author: LiuJie
+   * @Date: 2019/5/31
+   */
+    public void sendMessage(String message) throws IOException {
+        this.session.getBasicRemote().sendText(message);
+    }
+
+    /**
+     * @Description: 实现服务器主动向所有的客户端推送数据
+     * @Param:
+     * @return:
+     * @Author: LiuJie
+     * @Date: 2019/5/31
+     */
+    public void sendMessageToAll(Map<String,String> map) {
+        try {
+            for (WebSocketServer item : webSocketSet) {
+                item.session.getBasicRemote().sendText(JSON.toJSONString(map));
+            }
+            log.info("发送了。。。");
+        } catch (Exception e) {
+            String info = ExceptionMessage.getStackTraceInfo(e);
+            log.error("发生错误："+info);
+        }
+    }
+    
+    /** 
+    * @Description: 推送所有的报警信息到web端 
+    * @Param:  
+    * @return:  
+    * @Author: LiuJie 
+    * @Date: 2019/6/3 
+    */
+    public void sendMessageToAll() {
+//        SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd");
+//        String time = DateUtil.getStringDate(formatter);
+//        Calendar calendar = Calendar.getInstance();
+//        Date date=null;
+//        try {
+//            date = formatter.parse(time);
+//        } catch (Exception e) {
+//            e.printStackTrace();
+//        }
+//        calendar.setTime(date);
+//        Date startDate = calendar.getTime();
+//        int day=calendar.get(Calendar.DATE);
+//        calendar.set(Calendar.DATE,day+Properties.NUM_1);
+//        Date endDate = calendar.getTime();
+//        long startDateTime = startDate.getTime();
+//        long endDateTime = endDate.getTime();
+//        Map<String, Message> messageMap = dataContainer.getMessageMap();
+//        List<Message> messageList = new ArrayList<>();
+//        if (messageMap !=null && messageMap.size()>0){
+//            Iterator<String> iterator = messageMap.keySet().iterator();
+//            while (iterator.hasNext()){
+//                String key = iterator.next();
+////                log.info(" message key:{}",key);
+//                Message message = messageMap.get(key);
+//                long happenedTime = message.getHappenedTime().getTime();
+//                if (happenedTime>=startDateTime&&happenedTime<endDateTime){
+//                    messageList.add(message);
+//                }
+//            }
+//            Comparator comparator = getMessageComparator();
+//            Collections.sort(messageList,(e1,e2) -> comparator.compare(e1,e2) );
+//            Map<String, List<Message>> map = new HashMap<>();
+//            map.put(Properties.MESSAGES,messageList);
+//            String message = JSON.toJSONString(map);
+//            for (WebSocketServer item : webSocketSet) {
+//                try {
+//                    item.session.getBasicRemote().sendText(message);
+//                } catch (Exception e) {
+//                    String info = ExceptionMessage.getStackTraceInfo(e);
+//                    log.error("发生错误："+info);
+//                }
+//            }
+//        }
+
+    }
+
+
+    /**
+     * 群发自定义消息
+     * */
+    public void sendInfo(String message,@PathParam("sid") String sid) throws IOException {
+//        log.info("推送消息到窗口"+sid+"，推送内容:"+message);
+        for (WebSocketServer item : webSocketSet) {
+            try {
+                /**
+                 * 这里可以设定只推送给这个sid的，为null则全部推送
+                 */
+                if(sid==null) {
+                    item.sendMessage(message);
+                }else if(item.sid.equals(sid)){
+                    item.sendMessage(message);
+                }
+            } catch (IOException e) {
+                String info = ExceptionMessage.getStackTraceInfo(e);
+                log.error("error:"+info);
+                continue;
+            }
+        }
+    }
+
+    public static synchronized int getOnlineCount() {
+        return onlineCount;
+    }
+
+    public static synchronized void addOnlineCount() {
+        WebSocketServer.onlineCount++;
+    }
+
+    public static synchronized void subOnlineCount() {
+        WebSocketServer.onlineCount--;
+    }
+
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o){
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()){
+            return false;
+        }
+        if (!super.equals(o)){
+            return false;
+        }
+
+        WebSocketServer that = (WebSocketServer) o;
+
+        if (session != null ? !session.equals(that.session) : that.session != null){
+            return false;
+        }
+        return sid != null ? sid.equals(that.sid) : that.sid == null;
+    }
+
+    @Override
+    public int hashCode() {
+        int result = super.hashCode();
+        result = 31 * result + (session != null ? session.hashCode() : 0);
+        result = 31 * result + (sid != null ? sid.hashCode() : 0);
+        return result;
+    }
+
+}

src/main/java/com/zjty/inspect/controller/EvaluationController.java

@@ -146,7 +146,7 @@ public class EvaluationController {
     })
     @RequestMapping(value="/download",method= RequestMethod.POST)
     @AuthAnnotation(code = {"000800"})
-    public ResponseEntity findSearch(@RequestBody Download download){
+    public ResponseEntity download(@RequestBody Download download){
 //        return ResponseEntity.ok("http://120.55.57.35:12345/static/uplaods/a36b17568d4e466dacf9f088a29b4dbc.docx");
 //        return ResponseEntity.ok("http://120.55.57.35:8078/static/defult.docx");
         return ResponseEntity.ok("http://120.55.57.35:12345/static/uplaods/9800449e60d549568f4f0ddeb1974300.docx");

src/main/java/com/zjty/inspect/controller/WebSocketController.java

+package com.zjty.inspect.controller;
+
+import com.zjty.inspect.config.WebSocketServer;
+import com.zjty.inspect.thread.task.AsyncTask;
+import com.zjty.inspect.utils.ExceptionMessage;
+import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiOperation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.*;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * @program: liujie-cloud
+ * @description: WebSocketController，用于消息推送
+ * @author: LiuJie
+ * @create: 2019-05-17 10:12
+ **/
+
+@Api(description = "WebSocketController接口")
+@RequestMapping("/webSocket")
+@RestController
+public class WebSocketController {
+    /**
+     * 日志对象
+     */
+    private static final Logger logger = LoggerFactory.getLogger(WebSocketController.class);
+
+    @Autowired
+    private AsyncTask asyncTask;
+
+
+    /**
+     * @Description:  推送数据到所有的WebSocket客户端
+     * @Param:
+     * @return:
+     * @Author: LiuJie
+     * @Date: 2019/6/3
+     */
+    @ApiOperation(value = "推送数据到所有的WebSocket客户端" ,  notes="推送数据到所有的WebSocket客户端")
+    @RequestMapping(value="/push/{msg}",method= RequestMethod.GET)
+    public ResponseEntity pushToAll(@PathVariable("msg")String msg){
+        try {
+            Map<String, String> map = new HashMap<>();
+            map.put("msg",msg);
+            asyncTask.sendMessageToAll(map);
+        } catch (Exception e) {
+            String info = ExceptionMessage.getStackTraceInfo(e);
+            logger.error("推送消息到客户端报错:"+info);
+            return ResponseEntity.ok("推送消息到客户端报错:"+info);
+        }
+        return ResponseEntity.ok(200);
+    }
+}
\ No newline at end of file

src/main/java/com/zjty/inspect/thread/pool/TaskExecutePool.java

+package com.zjty.inspect.thread.pool;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.task.AsyncTaskExecutor;
+import org.springframework.scheduling.annotation.EnableAsync;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+
+import java.util.concurrent.ThreadPoolExecutor;
+
+/**
+ * @program: bserver-parent
+ * @description: TaskExecutePool,
+ *  @EnableAsync：开启异步处理
+ * @author: LiuJie
+ * @create: 2019-05-20 16:01
+ **/
+@Configuration
+@EnableAsync
+public class TaskExecutePool {
+    @Autowired
+    private ThreadPoolConfig config;
+
+    @Bean
+    public AsyncTaskExecutor myTaskAsyncPool() {
+        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+        /**
+         * 核心线程池大小
+         */
+        executor.setCorePoolSize(config.getCorePoolSize());
+        /**
+         * 最大线程数
+         */
+        executor.setMaxPoolSize(config.getMaxPoolSize());
+        /**
+         * 队列容量
+         */
+        executor.setQueueCapacity(config.getQueueCapacity());
+        /**
+         * 活跃时间
+         */
+        executor.setKeepAliveSeconds(config.getKeepAliveSeconds());
+        /**
+         * 线程名字前缀
+         */
+        executor.setThreadNamePrefix("MyExecutor-");
+
+        /**
+         * setRejectedExecutionHandler：当pool已经达到max size的时候，如何处理新任务
+         * CallerRunsPolicy：不在新线程中执行任务，而是由调用者所在的线程来执行
+         */
+        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
+        executor.initialize();
+        return executor;
+    }
+}
\ No newline at end of file

src/main/java/com/zjty/inspect/thread/pool/ThreadPoolConfig.java

+package com.zjty.inspect.thread.pool;
+
+import lombok.Data;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * @program: bserver-parent
+ * @description: 线程池属性配置类
+ * @author: LiuJie
+ * @create: 2019-05-20 16:01
+ **/
+@Data
+@Configuration
+//@ConfigurationProperties(prefix = "thread.pool")
+public class ThreadPoolConfig {
+    /**
+     * 核心线程池大小
+     */
+    @Value("${thread.pool.corePoolSize}")
+    private int corePoolSize;
+    /**
+     * 最大线程数
+     */
+    @Value("${thread.pool.maxPoolSize}")
+    private int maxPoolSize;
+    /**
+     * 活跃时间
+     */
+    @Value("${thread.pool.keepAliveSeconds}")
+    private int keepAliveSeconds;
+    /**
+     * 队列容量
+     */
+    @Value("${thread.pool.queueCapacity}")
+    private int queueCapacity;
+}

src/main/java/com/zjty/inspect/thread/task/AsyncTask.java

+package com.zjty.inspect.thread.task;
+
+import com.zjty.inspect.config.WebSocketServer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.scheduling.annotation.Async;
+import org.springframework.stereotype.Component;
+
+import java.util.*;
+
+@Component
+public class AsyncTask {
+    protected final Logger logger = LoggerFactory.getLogger(this.getClass());
+
+    @Autowired
+    private WebSocketServer webSocketServer;
+
+    /**
+    * @Description: myTaskAsynPool即配置线程池的方法名，
+     *  此处如果不写自定义线程池的方法名，会使用默认的线程池
+    * @Param:
+    * @return:
+    * @Author: LiuJie
+    * @Date: 2019/5/20
+    */
+    @Async("myTaskAsyncPool")
+    public void sendMessageToAll(Map<String,String> map) throws Exception{
+        webSocketServer.sendMessageToAll(map);
+    }
+
+}
\ No newline at end of file

src/main/java/com/zjty/inspect/thread/task/ScheduledService.java

+package com.zjty.inspect.thread.task;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+/**
+ * @program: bserver-parent
+ * @description: ScheduledService
+ * @author: LiuJie
+ * @create: 2019-05-20 15:36
+ **/
+
+@Slf4j
+@Component
+public class ScheduledService {
+
+    @Autowired
+    private AsyncTask asyncTask;
+
+    /**
+    * @Description: 每隔30秒，轮训消除一次未消除的报警消息
+    * @Param:
+    * @return:
+    * @Author: LiuJie
+    * @Date: 2019/5/31
+    */
+    @Scheduled(fixedRate = 1000 * 30)
+    public void removeMessage() {
+//        asyncTask.removeMessage();
+    }
+}
\ No newline at end of file

src/main/java/com/zjty/inspect/utils/ExceptionMessage.java

+package com.zjty.inspect.utils;
+
+import java.io.IOException;
+import java.io.PrintWriter;
+import java.io.StringWriter;
+
+/**
+ * @program: bserver
+ * @description: 异常信息捕获类，用于将异常信息转换为String
+ * @author: LiuJie
+ * @create: 2019-05-15 11:27
+ **/
+
+public class ExceptionMessage {
+    /**
+     * 获取e.printStackTrace() 的具体信息，赋值给String 变量，并返回
+     */
+    public static String getStackTraceInfo(Exception e) {
+        StringWriter sw = null;
+        PrintWriter pw = null;
+        try {
+            sw = new StringWriter();
+            pw = new PrintWriter(sw);
+            e.printStackTrace(pw);//将出错的栈信息输出到printWriter中
+            pw.flush();
+            sw.flush();
+            return sw.toString();
+        } catch (Exception ex) {
+            return "printStackTrace()转换错误";
+        } finally {
+            if (sw != null) {
+                try {
+                    sw.close();
+                } catch (IOException e1) {
+                    e1.printStackTrace();
+                }
+            }
+            if (pw != null) {
+                pw.close();
+            }
+        }
+
+    }
+
+    public static String getError(Throwable throwable) {
+        StringWriter sw = null;
+        PrintWriter pw = null;
+        try {
+            sw = new StringWriter();
+            pw = new PrintWriter(sw);
+            /**
+             * 将出错的栈信息输出到printWriter中
+             */
+            throwable.printStackTrace(pw);
+            pw.flush();
+            sw.flush();
+            return sw.toString();
+        } catch (Exception ex) {
+            return "printStackTrace()转换错误";
+        } finally {
+            if (sw != null) {
+                try {
+                    sw.close();
+                } catch (IOException e1) {
+                    e1.printStackTrace();
+                }
+            }
+            if (pw != null) {
+                pw.close();
+            }
+        }
+
+    }
+}

src/main/resources/application.properties

@@ -50,7 +50,8 @@ spring.resources.static-locations=classpath:/uploads/
 # mysql\u6570\u636E\u5E93\u914D\u7F6E
 spring.datasource.driver-class-name=com.mysql.jdbc.Driver
 #spring.datasource.url=jdbc:mysql://192.168.1.249:3306/bservice?useSSL=false&serverTimezone=UTC&autoReconnect=true&characterEncoding=utf-8
-spring.datasource.url=jdbc:mysql://localhost:3306/adaptation?useSSL=false&serverTimezone=UTC&autoReconnect=true&characterEncoding=utf-8
+#spring.datasource.url=jdbc:mysql://localhost:3306/adaptation?useSSL=false&serverTimezone=UTC&autoReconnect=true&characterEncoding=utf-8
+spring.datasource.url=jdbc:mysql://192.168.1.248:3306/adaptation?useSSL=false&serverTimezone=UTC&autoReconnect=true&characterEncoding=utf-8
 spring.datasource.username=root
 spring.datasource.password=root
 
@@ -67,3 +68,9 @@ spring.freemarker.template-loader-path=classpath:/templates
 ##Redis\u670D\u52A1\u5668\u8FDE\u63A5\u5BC6\u7801\uFF08\u9ED8\u8BA4\u4E3A\u7A7A\uFF09
 #spring.redis.password=
 address=120.55.57.35
+
+#   核心线程池大小、最大线程数、空闲活跃时间、队列容量
+thread.pool.corePoolSize=10
+thread.pool.maxPoolSize=40
+thread.pool.keepAliveSeconds=300
+thread.pool.queueCapacity=50
\ No newline at end of file