Springboot+websocket+定時器實現消息推送

由於最近有個需求，產品即將到期(不同時間段到期)時給后台用戶按角色推送，功能完成之后在此做個小結

1. 在啟動類中添加注解@EnableScheduling

package com.hsfw.backyard.websocket333;

/**
 * @Description
 * @Author: liucq
 * @Date: 2019/1/25
 */

import org.mybatis.spring.annotation.MapperScan;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.boot.web.support.SpringBootServletInitializer;
import org.springframework.scheduling.annotation.EnableScheduling;

@SpringBootApplication
@MapperScan("com.siwei.insurance.*.dao")
/**
 *  //該注解是開啟定時任務的支持
 */
@EnableScheduling
public class lifeInsuranceApplication extends SpringBootServletInitializer {
    @Override
    protected SpringApplicationBuilder configure(SpringApplicationBuilder builder) {
        return builder.sources(lifeInsuranceApplication.class);
    }

    public static void main(String[] args) {
        SpringApplication.run(lifeInsuranceApplication.class, args);
    }
}

2. 寫定時器

package com.hsfw.backyard.websocket333;

/**
 * @Description
 * @Author: liucq
 * @Date: 2019/1/25
 */

import com.siwei.insurance.permission.dao.RolePermissionDao;
import com.siwei.insurance.productManage.dao.ExpirePushMsgMapper;
import com.siwei.insurance.productManage.service.ProductService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;

import java.util.ArrayList;
import java.util.List;

@Componentpublic
class ProductExpireTask {
    @Autowired
    private RolePermissionDao rolePermissionDao;
    @Autowired
    private ProductService productService;
    @Autowired
    private ExpirePushMsgMapper expirePushMsgMapper;

    /**
     * 每天早上0點執行
     */

    @Scheduled(cron = "0 0 0 1/1 * ?")
    public void productExpire() {
        //距離到期還有一個月提醒
        String oneMonthExpireDate = DateUtil.addOneMonth();
        dealExpireProduct(oneMonthExpireDate);
        //距離到期還有一天提醒
        String oneDayExpireDate = DateUtil.addOneDay();
        dealExpireProduct(oneDayExpireDate);
        //距離到期還有一周提醒
        String oneWeekExpireDate = DateUtil.addFewDays(7);
        dealExpireProduct(oneWeekExpireDate);
    }

    private void dealExpireProduct(String expireDate) {
        List<Map<String, Object>> expireProductMapList = productService.findExpireProducts(expireDate);
        if (expireProductMapList != null && !expireProductMapList.isEmpty()) {
            // 根據路徑查詢需要推送的角色
            List<String> needPushRoleIds = rolePermissionDao.findNeedPushRoleIdByUrl(TotalConstant.PRODUCT_PUSH_URL);
            List<ExpirePushMsg> expirePushMsgs = new ArrayList<>();
            for (Map<String, Object> expireProductMap : expireProductMapList) {
                ExpirePushMsg expirePushMsg = new ExpirePushMsg();
                expirePushMsg.setNeedPushEntityId((int) expireProductMap.get("id"));
                expirePushMsg.setNeedPushEntityNo((String) expireProductMap.get("insuranceNo"));
                String productName = (String) expireProductMap.get("insuranceName");
                expirePushMsg.setNeedPushEntityName(productName);
                //設置此推送消息的到期時間
                expirePushMsg.setExpireDate(DateUtil.stringToDate(DateUtil.addOneDay()));
                expirePushMsg.setPushType(2);
                StringBuffer needPushRoleIdString = new StringBuffer();
                needPushRoleIds.forEach(e -> needPushRoleIdString.append(e + ";"));
                expirePushMsg.setPushRoleId(needPushRoleIdString.toString().trim());
                String productExpireDateString = DateUtil.dateToShotString((Date) expireProductMap.get("expiryDate"));
                expirePushMsg.setPushMsg("您的產品：" + productName + "，將於" + productExpireDateString + "即將過期，請及時處理！");
                expirePushMsgs.add(expirePushMsg);
            }
            expirePushMsgMapper.insertAll(expirePushMsgs);
        }
    }

    @Scheduled(cron = "0 0 0 1/1 * ?")
    public void pushMsgExpire() {
        String oneDayExpireDate = DateUtil.getZeroTime(DateUtil.addOneDay());
        //推送消息只存在一天，根據到期時間將數據刪除
        expirePushMsgMapper.deleteByExpireDate(oneDayExpireDate);
    }
}

DateUtil工具類

package com.hsfw.backyard.websocket333;

/**
 * @Description
 * @Author: liucq
 * @Date: 2019/1/25
 */

import org.apache.commons.lang3.StringUtils;

import java.text.ParsePosition;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;

/**
 * @Description 時間處理工具類  * @author linxiunan * @date 2018年9月3日
 */
public class DateUtil {
    private static final SimpleDateFormat dayOfDateFormat = new SimpleDateFormat("yyyy-MM-dd");
    private static final SimpleDateFormat secondOfDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    /**
     * @return 當天時間加一天，返回"yyyy-MM-dd"格式
     */
    public static String addOneDay() {
        Calendar calendar = Calendar.getInstance();
        calendar.add(Calendar.DAY_OF_MONTH, 1);
        return dayOfDateFormat.format(calendar.getTime());
    }

    /**
     * @return 當天時間加一月，返回"yyyy-MM-dd"格式
     */
    public static String addOneMonth() {
        Calendar calendar = Calendar.getInstance();
        calendar.add(Calendar.MONTH, 1);
        return dayOfDateFormat.format(calendar.getTime());
    }

    /**
     * @param dayNumber 加的天數     * @return 返回當天時間添加幾天之后的時間,返回"yyyy-MM-dd"格式
     */
    public static String addFewDays(int dayNumber) {
        Calendar calendar = Calendar.getInstance();
        calendar.add(Calendar.DAY_OF_MONTH, dayNumber);
        return dayOfDateFormat.format(calendar.getTime());
    }

    /**
     * @param dateString 需要轉換成時間格式的日期字符串     * @return 返回字符串轉換成的時間
     */
    public static Date stringToDate(String dateString) {
        ParsePosition parsePosition = new ParsePosition(0);
        if (dateString.contains(" ")) {
            return secondOfDateFormat.parse(dateString, parsePosition);
        } else {
            return dayOfDateFormat.parse(dateString, parsePosition);
        }
    }

    /**
     * @param date 需要轉換成字符串格式的日期     * @return 返回"yyyy-MM-dd"格式的轉換后的字符串
     */
    public static String dateToShotString(Date date) {
        return dayOfDateFormat.format(date);
    }

    /**
     * @param date 需要轉換成字符串格式的日期     * @return 返回"yyyy-MM-dd HH:mm:ss"格式的轉換后的字符串
     */
    public static String dateToLongString(Date date) {
        return secondOfDateFormat.format(date);
    }

    /**
     * @param dateString 需要獲取0點的時間字符串,如果獲取當天0點,傳null即可     * @return 返回"yyyy-MM-dd HH:mm:ss"格式的某天0點字符串
     */
    public static String getZeroTime(String dateString) {
        if (StringUtils.isBlank(dateString)) {
            Calendar calendar = Calendar.getInstance();
            calendar.set(Calendar.HOUR_OF_DAY, 0);
            calendar.set(Calendar.MINUTE, 0);
            calendar.set(Calendar.SECOND, 0);
            return secondOfDateFormat.format(calendar.getTime());
        } else {
            Date date = stringToDate(dateString);
            return dateToLongString(date);
        }
    }
}

3. 引入websocket所需jar包

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-websocket</artifactId>
</dependency>

4. 配置websocket

編寫MyEndpointConfigure類

package com.hsfw.backyard.websocket333;

/**
 * @Description
 * @Author: liucq
 * @Date: 2019/1/25
 */

import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;

import javax.websocket.server.ServerEndpointConfig;

public class MyEndpointConfigure extends ServerEndpointConfig.Configurator implements ApplicationContextAware {
    private static volatile BeanFactory context;

    @Override
    public <T> T getEndpointInstance(Class<T> clazz) throws InstantiationException {
        return context.getBean(clazz);
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        MyEndpointConfigure.context = applicationContext;
    }
}

websocket配置類

package com.hsfw.backyard.websocket333;

/**
 * @Description
 * @Author: liucq
 * @Date: 2019/1/25
 */

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;

@Configuration
public class WebSocketConfig {
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }

    @Bean
    public MyEndpointConfigure newConfigure() {
        return new MyEndpointConfigure();
    }
}

這里需要重點說明一下，在websocket配置類中，第一個配置是因為使用springboot內置容器，自己開發時需要配置，如果有獨立的容器需要將其注釋掉，也就意味着，如果將項目打成WAR包，部署到服務器，使用Tomcat啟動時，需要注釋掉ServerEndpointExporter配置；MyEndpointConfigure配置是因為我的需求需要，需要在websocket類中注入service層或者dao層的接口，MyEndpointConfigure配置就是為了解決websocket無法注入的問題，如果沒有需要可以不用配置
---------------------

5. websocket類

package com.hsfw.backyard.websocket333;

/**
 * @Description
 * @Author: liucq
 * @Date: 2019/1/25
 */

import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.websocket.OnClose;
import javax.websocket.OnError;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CopyOnWriteArraySet;

@Component
@ServerEndpoint(value = "/productWebSocket/{userId}", configurator = MyEndpointConfigure.class)
public class ProductWebSocket {
    // 靜態變量，用來記錄當前在線連接數。應該把它設計成線程安全的。
    private static int onlineCount = 0;
    // concurrent包的線程安全Set，用來存放每個客戶端對應的ProductWebSocket對象。
    private static CopyOnWriteArraySet<ProductWebSocket> webSocketSet = new CopyOnWriteArraySet<ProductWebSocket>();
    // 與某個客戶端的連接會話，需要通過它來給客戶端發送數據    private Session session;
    @Autowired
    private UserRoleDao userRoleDao;
    @Autowired
    private ExpirePushMsgMapper expirePushMsgMapper;
    private Logger log = LoggerFactory.getLogger(ProductWebSocket.class);

    /**
     * 連接建立成功調用的方法
     */
    @OnOpen
    public void onOpen(@PathParam("userId") String userId, Session session) {
        log.info("新客戶端連入，用戶id：" + userId);
        this.session = session;
        webSocketSet.add(this); // 加入set中
        addOnlineCount(); // 在線數加1
        // 相關業務處理，根據拿到的用戶ID判斷其為那種角色，根據角色ID去查詢是否有需要推送給該角色的消息，有則推送
        if (StringUtils.isNotBlank(userId)) {
            List<String> roleIds = userRoleDao.findRoleIdByUserId(userId);
            List<String> totalPushMsgs = new ArrayList<String>();
            for (String roleId : roleIds) {
                List<String> pushMsgs = expirePushMsgMapper.findPushMsgByRoleId(roleId);
                if (pushMsgs != null && !pushMsgs.isEmpty()) {
                    totalPushMsgs.addAll(pushMsgs);
                }
            }
            if (totalPushMsgs != null && !totalPushMsgs.isEmpty()) {
                totalPushMsgs.forEach(e -> sendMessage(e));
            }
        }
    }

    /**
     * 連接關閉調用的方法
     */
    @OnClose
    public void onClose() {
        log.info("一個客戶端關閉連接");
        webSocketSet.remove(this); // 從set中刪除
        subOnlineCount(); // 在線數減1
    }

    /**
     * 收到客戶端消息后調用的方法
     * * @param message     客戶端發送過來的消息
     */
    @OnMessage
    public void onMessage(String message, Session session) {
    }

    /**
     * 發生錯誤時調用
     */
    @OnError
    public void onError(Session session, Throwable error) {
        log.error("websocket出現錯誤");
        error.printStackTrace();
    }

    public void sendMessage(String message) {
        try {
            this.session.getBasicRemote().sendText(message);
            log.info("推送消息成功，消息為：" + message);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 群發自定義消息
     */
    public static void sendInfo(String message) throws IOException {
        for (ProductWebSocket productWebSocket : webSocketSet) {
            productWebSocket.sendMessage(message);
        }
    }

    public static synchronized int getOnlineCount() {
        return onlineCount;
    }

    public static synchronized void addOnlineCount() {
        ProductWebSocket.onlineCount++;
    }

    public static synchronized void subOnlineCount() {
        ProductWebSocket.onlineCount--;
    }
}

這樣后台的功能基本上就算是寫完了，前端配合測試一下

6. 前端測試

寫一個頁面，代碼如下

<!DOCTYPE HTML>
<html>
<head>
    <title>My WebSocket</title>
</head>
 
 
<body>
Welcome<br/>
<input id="text" type="text" /><button onclick="send()">Send</button>    <button onclick="closeWebSocket()">Close</button>
<div id="message">
</div>
</body>
 
 
<script type="text/javascript">
    var websocket = null;
 
 
    //判斷當前瀏覽器是否支持WebSocket
    if('WebSocket' in window){
        websocket = new WebSocket("ws://localhost:8080/productWebSocket/001");
    }
    else{
        alert('Not support websocket')
    }
 
 
    //連接發生錯誤的回調方法
    websocket.onerror = function(){
        setMessageInnerHTML("error");
    };
 
 
    //連接成功建立的回調方法
    websocket.onopen = function(event){
        setMessageInnerHTML("open");
    }
 
 
    //接收到消息的回調方法
    websocket.onmessage = function(event){
        setMessageInnerHTML(event.data);
    }
 
 
    //連接關閉的回調方法
    websocket.onclose = function(){
        setMessageInnerHTML("close");
    }
 
 
    //監聽窗口關閉事件，當窗口關閉時，主動去關閉websocket連接，防止連接還沒斷開就關閉窗口，server端會拋異常。
    window.onbeforeunload = function(){
        websocket.close();
    }
 
 
    //將消息顯示在網頁上
    function setMessageInnerHTML(innerHTML){
        document.getElementById('message').innerHTML += innerHTML + '<br/>';
    }
 
 
    //關閉連接
    function closeWebSocket(){
        websocket.close();
    }
 
 
    //發送消息
    function send(){
        var message = document.getElementById('text').value;
        websocket.send(message);
    }
</script>
</html>

項目啟動之后，打開頁面

open下方就是我添加的消息，可以看出已經成功推送，到此該功能就算完成結束了

參考地址：https://www.cnblogs.com/bianzy/p/5822426.html