Springboot+websocket+定時器實現訊息推送
阿新 • • 發佈：2019-01-05
由於最近有個需求，產品即將到期(不同時間段到期)時給後臺使用者按角色推送，功能完成之後在此做個小結
1. 在啟動類中添加註解@EnableScheduling
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. 寫定時器
import java.util.ArrayList; import java.util.Date; import java.util.List; import java.util.Map; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; import com.siwei.insurance.entity.ExpirePushMsg; import com.siwei.insurance.helper.TotalConstant; import com.siwei.insurance.permission.dao.RolePermissionDao; import com.siwei.insurance.productManage.dao.ExpirePushMsgMapper; import com.siwei.insurance.productManage.service.ProductService; import com.siwei.insurance.until.DateUtil; @Component public 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.siwei.insurance.until; import java.text.ParsePosition; import java.text.SimpleDateFormat; import java.util.Calendar; import java.util.Date; import org.apache.commons.lang3.StringUtils; /** * @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類
import javax.websocket.server.ServerEndpointConfig;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
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配置類
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;
import com.siwei.insurance.websocket.MyEndpointConfigure;
@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類
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CopyOnWriteArraySet;
import javax.websocket.OnClose;
import javax.websocket.OnError;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import com.siwei.insurance.permission.dao.UserRoleDao;
import com.siwei.insurance.productManage.dao.ExpirePushMsgMapper;
@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下方就是我新增的訊息，可以看出已經成功推送，到此該功能就算完成結束了