背景
最近公司的客戶要求,分配給員工的任務除了有微信通知外,還希望PC端的網頁也能實時收到通知。管理員分配任務是在我們的系統A,而員工接受任務是在系統B。兩個系統都是現在已投入使用的系統。
技術選型
根據需求我們最終選用SpringAOP+RabbitMQ+WebSocket。
SpringAOP可以讓我們不修改原有代碼,直接將原有service作為切點,加入切面。RabbitMQ可以讓A系統和B系統解耦。WebSocket則可以達到實時通知的要求。
SpringAOP
AOP稱為面向切面編程,在程序開發中主要用來解決一些系統層面上的問題,比如日志,事務,權限等待。是Spring的核心模塊,底層是通過動態代理來實現(動態代理將在之后的文章重點介紹)。
基本概念
Aspect(切面):通常是一個類,里面可以定義切入點和通知。
JointPoint(連接點):程序執行過程中明確的點,一般是方法的調用。
Advice(通知):AOP在特定的切入點上執行的增強處理,有before,after,afterReturning,afterThrowing,around。
Pointcut(切入點):就是帶有通知的連接點,在程序中主要體現為書寫切入點表達式。
通知類型
Before:在目標方法被調用之前做增強處理。
@Before只需要指定切入點表達式即可
AfterReturning:在目標方法正常完成后做增強。
@AfterReturning除了指定切入點表達式后,還可以指定一個返回值形參名returning,代表目標方法的返回值
AfterThrowing:主要用來處理程序中未處理的異常。
@AfterThrowing除了指定切入點表達式后,還可以指定一個throwing的返回值形參名,可以通過該形參名
來訪問目標方法中所拋出的異常對象
After:在目標方法完成之后做增強,無論目標方法時候成功完成。
@After可以指定一個切入點表達式
Around:環繞通知,在目標方法完成前后做增強處理,環繞通知是最重要的通知類型,像事務,日志等都是環繞通知,注意編程中核心是一個ProceedingJoinPoint。
RabbitMQ
(圖摘自:https://www.cnblogs.com/dwlsxj/p/RabbitMQ.html)
從圖中我們可以看到RabbitMQ主要的結構有:Routing、Binding、Exchange、Queue。
Queue
Queue(隊列)RabbitMQ的作用是存儲消息,隊列的特性是先進先出。
Exchange
生產者產生的消息並不是直接發送給消息隊列Queue的,而是要經過Exchange(交換器),由Exchange再將消息路由到一個或多個Queue,還會將不符合路由規則的消息丟棄。
Routing
用於標記或生產者尋找Exchange。
Binding
用於Exchange和Queue做關聯。
Exchange Type
fanout
fanout類型的Exchange路由規則非常簡單,它會把所有發送到該Exchange的消息路由到所有與它綁定的Queue中。
direct
direct會把消息路由到那些binding key與routing key完全匹配的Queue中。
topic
direct規則是嚴格意義上的匹配,換言之Routing Key必須與Binding Key相匹配的時候才將消息傳送給Queue,那么topic這個規則就是模糊匹配,可以通過通配符滿足一部分規則就可以傳送。
headers
headers類型的Exchange不依賴於routing key與binding key的匹配規則來路由消息,而是根據發送的消息內容中的headers屬性進行匹配。
WebSocket
了解websocket必須先知道幾個常用的web通信技術及其區別。
短輪詢
短輪詢的基本思路就是瀏覽器每隔一段時間向瀏覽器發送http請求,服務器端在收到請求后,不論是否有數據更新,都直接進行響應。這種方式實現的即時通信,本質上還是瀏覽器發送請求,服務器接受請求的一個過程,通過讓客戶端不斷的進行請求,使得客戶端能夠模擬實時地收到服務器端的數據的變化。
這種方式的優點是比較簡單,易於理解,實現起來也沒有什么技術難點。缺點是顯而易見的,這種方式由於需要不斷的建立http連接,嚴重浪費了服務器端和客戶端的資源。尤其是在客戶端,距離來說,如果有數量級想對比較大的人同時位於基於短輪詢的應用中,那么每一個用戶的客戶端都會瘋狂的向服務器端發送http請求,而且不會間斷。人數越多,服務器端壓力越大,這是很不合理的。
因此短輪詢不適用於那些同時在線用戶數量比較大,並且很注重性能的Web應用。
長輪詢/comet
comet指的是,當服務器收到客戶端發來的請求后,不會直接進行響應,而是先將這個請求掛起,然后判斷服務器端數據是否有更新。如果有更新,則進行響應,如果一直沒有數據,則到達一定的時間限制(服務器端設置)后關閉連接。
長輪詢和短輪詢比起來,明顯減少了很多不必要的http請求次數,相比之下節約了資源。長輪詢的缺點在於,連接掛起也會導致資源的浪費。
SSE
SSE是HTML5新增的功能,全稱為Server-Sent Events。它可以允許服務推送數據到客戶端。SSE在本質上就與之前的長輪詢、短輪詢不同,雖然都是基於http協議的,但是輪詢需要客戶端先發送請求。而SSE最大的特點就是不需要客戶端發送請求,可以實現只要服務器端數據有更新,就可以馬上發送到客戶端。
SSE的優勢很明顯,它不需要建立或保持大量的客戶端發往服務器端的請求,節約了很多資源,提升應用性能。並且SSE的實現非常簡單,不需要依賴其他插件。
WebSocket
WebSocket是Html5定義的一個新協議,與傳統的http協議不同,該協議可以實現服務器與客戶端之間全雙工通信。簡單來說,首先需要在客戶端和服務器端建立起一個連接,這部分需要http。連接一旦建立,客戶端和服務器端就處於平等的地位,可以相互發送數據,不存在請求和響應的區別。
WebSocket的優點是實現了雙向通信,缺點是服務器端的邏輯非常復雜。現在針對不同的后台語言有不同的插件可以使用。
四種Web即時通信技術比較
從兼容性角度考慮,短輪詢>長輪詢>長連接SSE>WebSocket;
從性能方面考慮,WebSocket>長連接SSE>長輪詢>短輪詢。
實戰
項目使用SpringBoot搭建。RabbitMQ的安裝這里不講述。
RabbitMQ配置
兩個系統A、B都需要操作RabbitMQ,其中A生產消息,B消費消息。故都需要配置。
1、首先引入RabbitMQ的dependency:
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency>
這個dependency中包含了RabbitMQ相關dependency。
2、在項目的配置文件里配置為使用rabbitmq及其參數。
application-pro.yml
#消息隊列 message.queue.type: rabbitmq ## rabbit mq properties rabbitmq: host: localhost port: 5672 username: guest password: guest
application.properties
#將要使用的隊列名 rabbitmq.websocket.msg.queue=websocket_msg_queue
3、創建配置文件。隊列的創建交給spring。
RabbitMQConfig.java
@Configuration @EnableRabbit public class RabbitMQConfig { @Value("${rabbitmq.host}") private String host; @Value("${rabbitmq.port}") private String port; @Value("${rabbitmq.username}") private String username; @Value("${rabbitmq.password}") private String password; @Value("${rabbitmq.websocket.msg.queue}") private String webSocketMsgQueue; @Bean public ConnectionFactory connectionFactory() throws IOException { CachingConnectionFactory factory = new CachingConnectionFactory(); factory.setUsername(username); factory.setPassword(password); // factory.setVirtualHost("test"); factory.setHost(host); factory.setPort(Integer.valueOf(port)); factory.setPublisherConfirms(true); //設置隊列參數,是否持久化、隊列TTL、隊列消息TTL等 factory.createConnection().createChannel(false).queueDeclare(webSocketMsgQueue, true, false, false, null); return factory; } @Bean public MessageConverter messageConverter() { return new Jackson2JsonMessageConverter(); } @Bean @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE) // 必須是prototype類型 public RabbitTemplate rabbitTemplate() throws IOException { return new RabbitTemplate(connectionFactory()); } @Bean public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() throws IOException { SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory(); factory.setConnectionFactory(connectionFactory()); factory.setConcurrentConsumers(3); factory.setMaxConcurrentConsumers(10); factory.setAcknowledgeMode(AcknowledgeMode.MANUAL); return factory; } }
4、系統B中創建隊列監聽,當隊列有消息時,發送websocket通知。
RabbitMQListener.java
@Component public class RabbitMQListener { @Autowired private RabbitMQService mqService; /** * WebSocket推送監聽器 * @param socketEntity * @param deliveryTag * @param channel */ @RabbitListener(queues = "websocket_msg_queue") public void webSocketMsgListener(@Payload WebSocketMsgEntity socketMsgEntity, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Channel channel) throws IOException { mqService.handleWebSocketMsg(socketMsgEntity, deliveryTag, channel); } }
RabbitMQService.java
public class RabbitMQService { @Autowired private MessageWebSocketHandler messageWebSocketHandler; /** * @param socketMsgEntity * @param deliveryTag * @param channel * @throws IOException */ void handleWebSocketMsg(WebSocketMsgEntity socketMsgEntity, long deliveryTag, Channel channel) throws IOException { try { messageWebSocketHandler.sendMessageToUsers(socketMsgEntity.toJsonString(), socketMsgEntity.getToUserIds()); channel.basicAck(deliveryTag, false); } catch (Exception e) { channel.basicNack(deliveryTag, false, false); } } }
WebSocketMsgEntity為MQ中傳送的實體。
public class WebSocketMsgEntity implements Serializable { public enum OrderType{ repair("維修"), maintain("保養"), measure("計量"); OrderType(String value){ this.value = value; } String value; public String getValue() { return value; } } //設備名稱 private String EquName; //設備編號 private String EquId; //工單類型 private OrderType orderType; //工單單號 private String orderId; //工單狀態 private String orderStatus; //創建時間 private Date createTime; //消息接收人ID private List<String> toUserIds; public String getEquName() { return EquName; } public void setEquName(String equName) { EquName = equName; } public String getOrderId() { return orderId; } public void setOrderId(String orderId) { this.orderId = orderId; } public String getEquId() { return EquId; } public void setEquId(String equId) { EquId = equId; } public String getOrderStatus() { return orderStatus; } public void setOrderStatus(String orderStatus) { this.orderStatus = orderStatus; } public OrderType getOrderType() { return orderType; } public void setOrderType(OrderType orderType) { this.orderType = orderType; } public Date getCreateTime() { return createTime; } public void setCreateTime(Date createTime) { this.createTime = createTime; } public List<String> getToUserIds() { return toUserIds; } public void setToUserIds(List<String> toUserIds) { this.toUserIds = toUserIds; } public String toJsonString(){ return JSON.toJSONString(this); } }
SpringAOP
1、系統A中創建一個切面類DataInterceptor.java
@Aspect @Component public class DataInterceptor { @Autowired private MessageQueueService queueService; //維修工單切點 @Pointcut("execution(* com.zhishang.hes.common.service.impl.RepairServiceImpl.executeFlow(..))") private void repairMsg() { } /** * 返回通知,方法執行正常返回時觸發 * * @param joinPoint * @param result */ @AfterReturning(value = "repairMsg()", returning = "result") public void afterReturning(JoinPoint joinPoint, Object result) { //此處可以獲得切點方法名 //String methodName = joinPoint.getSignature().getName(); EquipmentRepair equipmentRepair = (EquipmentRepair) result; WebSocketMsgEntity webSocketMsgEntity = this.generateRepairMsgEntity(equipmentRepair); if (webSocketMsgEntity == null) { return; } queueService.send(webSocketMsgEntity); } /** * 生成發送到MQ的維修消息 * * @param equipmentRepair * @return */ private WebSocketMsgEntity generateRepairMsgEntity(EquipmentRepair equipmentRepair) { WebSocketMsgEntity webSocketMsgEntity = generateRepairMsgFromTasks(equipmentRepair); return webSocketMsgEntity; } /** * 從任務中生成消息 * * @param equipmentRepair * @return */ private WebSocketMsgEntity generateRepairMsgFromTasks(EquipmentRepair equipmentRepair) { //業務代碼略 } }
2、發送消息到MQ。這里只貼了發送的核心代碼
public class RabbitMessageQueue extends AbstractMessageQueue { @Value("${rabbitmq.websocket.msg.queue}") private String webSocketMsgQueue; @Autowired private RabbitTemplate rabbitTemplate; @Override public void send(WebSocketMsgEntity entity) { //沒有指定exchange,則使用默認名為“”的exchange,binding名與queue名相同 rabbitTemplate.convertAndSend(webSocketMsgQueue, entity); } }
WebSocket
1、 系統B中引入websocket服務端dependency
<dependency> <groupId>org.springframework</groupId> <artifactId>spring-websocket</artifactId> <version>4.3.10.RELEASE</version> </dependency>
2、 配置websocket,添加處理類
WebSocketConfigurer.java
@Configuration @EnableWebSocket public class WebSocketConfig extends WebMvcConfigurerAdapter implements WebSocketConfigurer { private static Logger logger = LoggerFactory.getLogger(WebSocketConfig.class); @Override public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) { //配置webSocket路徑 registry.addHandler(messageWebSocketHandler(),"/msg-websocket").addInterceptors(new MyHandshakeInterceptor()).setAllowedOrigins("*"); //配置webSocket路徑 支持前端使用socketJs registry.addHandler(messageWebSocketHandler(), "/sockjs/msg-websocket").setAllowedOrigins("*").addInterceptors(new MyHandshakeInterceptor()).withSockJS(); } @Bean public MessageWebSocketHandler messageWebSocketHandler() { logger.info("......創建MessageWebSocketHandler......"); return new MessageWebSocketHandler(); } }
MessageWebSocketHandler.java 主要用於websocket連接及消息發送處理。配置中還使用了連接握手時的處理,主要是取用戶登陸信息,這里不多講述。
public class MessageWebSocketHandler extends TextWebSocketHandler { private static Logger logger = LoggerFactory.getLogger(SystemWebSocketHandler.class); private static ConcurrentHashMap<String, CopyOnWriteArraySet<WebSocketSession>> users = new ConcurrentHashMap<>(); @Override public void afterConnectionEstablished(WebSocketSession session) throws Exception { String userId = session.getAttributes().get("WEBSOCKET_USERID").toString(); logger.info("......AfterConnectionEstablished......"); logger.info("session.getId:" + session.getId()); logger.info("session.getLocalAddress:" + session.getLocalAddress().toString()); logger.info("userId:" + userId); //websocket連接后記錄連接信息 if (users.keySet().contains(userId)) { CopyOnWriteArraySet<WebSocketSession> webSocketSessions = users.get(userId); webSocketSessions.add(session); } else { CopyOnWriteArraySet<WebSocketSession> webSocketSessions = new CopyOnWriteArraySet<>(); webSocketSessions.add(session); users.put(userId, webSocketSessions); } } @Override public void handleTransportError(WebSocketSession session, Throwable throwable) throws Exception { removeUserSession(session); if (session.isOpen()) { session.close(); } logger.info("異常出現handleTransportError" + throwable.getMessage()); } @Override public void afterConnectionClosed(WebSocketSession session, CloseStatus closeStatus) throws Exception { removeUserSession(session); logger.info("關閉afterConnectionClosed" + closeStatus.getReason()); } @Override public boolean supportsPartialMessages() { return false; } /** * 給符合要求的在線用戶發送消息 * * @param message */ public void sendMessageToUsers(String message, List<String> userIds) throws IOException{ if (StringUtils.isEmpty(message) || CollectionUtils.isEmpty(userIds)) { return; } if (users.isEmpty()) { return; } for (String userId : userIds) { if (!users.keySet().contains(userId)) { continue; } CopyOnWriteArraySet<WebSocketSession> webSocketSessions = users.get(userId); if (webSocketSessions == null) { continue; } for (WebSocketSession webSocketSession : webSocketSessions) { if (webSocketSession.isOpen()) { try { webSocketSession.sendMessage(new TextMessage(message)); } catch (IOException e) { logger.error(" WebSocket server send message ERROR " + e.getMessage()); try { throw e; } catch (IOException e1) { e1.printStackTrace(); } } } } } } /** * websocket清除連接信息 * * @param session */ private void removeUserSession(WebSocketSession session) { String userId = session.getAttributes().get("WEBSOCKET_USERID").toString(); if (users.keySet().contains(userId)) { CopyOnWriteArraySet<WebSocketSession> webSocketSessions = users.get(userId); webSocketSessions.remove(session); if (webSocketSessions.isEmpty()) { users.remove(userId); } } } }
整個功能完成后,A系統分配任務時,系統B登陸用戶收到的消息如圖:
總體流程:
1、對於系統B,每個登陸的用戶都會和服務器建立websocket長連接。
2、系統A生成任務,AOP做出響應,將封裝的消息發送給MQ。
3、系統B中的MQ監聽發現隊列有消息到達,消費消息。
4、系統B通過websocket長連接將消息發給指定的登陸用戶。
參考:
https://docs.spring.io/spring/docs/4.3.12.RELEASE/spring-framework-reference/htmlsingle/#websocket
https://www.cnblogs.com/jingmoxukong/p/7755643.html
https://www.cnblogs.com/dwlsxj/p/RabbitMQ.html
https://blog.csdn.net/Holmofy/article/details/78111715