這里只實現服務器端WebScket到消息中間件RabbitMQ部分，前端代碼不會。前端跟中間件交互部分的功能（向中間件發送消息、從中間件讀取消息）用接口代替
實現思路
前端發起請求與服務器建立連接 ->WebSocket發送消息到RabbitMQ隊列中->WebSocket監聽Rabbit消息隊列中的消息
項目結構
配置RabbitMQ
#配置rabbitmq的基本信息 : ip 端口 賬號和密碼
spring:
rabbitmq:
host: localhost
port: 5672
username: guest
password: guest
virtual-host: /
寫一個RabbitMQ的配置類
package com.example.demo.config;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* @author lyd
* @Description: 配置RabbitMQ，創建一個交換機、一個隊列
* @date 14:46
*/
@Configuration
public class AppConfig {
public static final String ROUTING_KEY = "rabbit.msg";
public static final String DIRECT_EXCHANGE = "directexchange";
public static final String DIRECT_QUEUE = "directqueue";
@Bean
public Queue directQueue() {
return new Queue(DIRECT_QUEUE);
}
@Bean
public DirectExchange directExchange() {
return new DirectExchange(DIRECT_EXCHANGE);
}
@Bean
public Binding binding() {
return BindingBuilder.bind(directQueue()).to(directExchange()).with(ROUTING_KEY);
}
}
再寫一個WebSocket啟動類，開啟WebSocket支持
package com.example.demo.config;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;
/**
* @author lyd
* @Description: 開啟WebSocket支持
* @date 15:43
*/
@Configuration
@Component
public class WebSocketConfig {
@Bean
public ServerEndpointExporter serverEndpointExporter(){
return new ServerEndpointExporter();
}
}
實現WebSocket的核心類，通過@OnOpen、@OnClose、@OnMessag、@OnError四個注解實現四個核心的方法
package com.example.demo.websocket;
import com.example.demo.direct.DirectSender;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.concurrent.CopyOnWriteArraySet;
@ServerEndpoint(value = "/webSocket")
@Component
public class WebSocketServer {
/**
* 存放每個客戶端對應的WebSocket對象
*/
public static CopyOnWriteArraySet<WebSocketServer> webSockets = new CopyOnWriteArraySet<WebSocketServer>();
@Autowired
static DirectSender directSender;
/**
* 連接建立成功調用的方法
*/
@OnOpen
public void onOpen() throws InterruptedException, IOException {
webSockets.add(this);
System.out.println("有新用戶加入");
}
@OnClose
public void onClose() throws IOException {
webSockets.remove(this);
System.out.println("有用戶離開");
}
/**
* 收到客戶端消息后調用的方法
*/
@OnMessage
public void onMessage(String msg) throws InterruptedException {
System.out.println("從客戶端接受的消息： " + msg);
}
@OnError
public void onError(Throwable error) {
error.printStackTrace();
}
}
定義一個消息發送類，實現一個通過WebSocket向隊列發送消息的方法
package com.example.demo.direct;
import com.example.demo.config.AppConfig;
import com.example.demo.websocket.WebSocketServer;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
/**
* @author lyd
* @Description:
* @date 14:50
*/
@Component
public class DirectSender {
@Autowired
RabbitTemplate rabbitTemplate;
/**
* 向RabbitMQ隊列中發送消息，方便后面客戶端可以從隊列中讀取該消息
*
* 也可以用來代替客戶端向隊列中發送消息，我不會寫前端連接rabbitmq的代碼，就用這個接口代替了。或者在RabbitMQ的管理面板中手動輸入數據
* @param msg
*/
public void sendDirect(String msg) {
for (WebSocketServer webSocketServer : WebSocketServer.webSockets) {
rabbitTemplate.convertAndSend(AppConfig.DIRECT_EXCHANGE, AppConfig.ROUTING_KEY, msg +" ("+webSocketServer.toString()+")");
}
}
}
定義一個消息接收類，實現一個通過WebSocket從隊列中讀取消息的方法
package com.example.demo.direct;
import com.example.demo.config.AppConfig;
import com.example.demo.websocket.WebSocketServer;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.io.IOException;
/**
* @author lyd
* @Description: 監聽RabbitMQ中隊列消息
* @date 15:16
*/
@Component
public class DirectReceive {
/**
* 監聽客戶端發送到RabbitMQ隊列中的消息，並把消息發送給WebSocketServer
* @param msg
* @throws InterruptedException
* @throws IOException
*/
@RabbitListener(queues = AppConfig.DIRECT_QUEUE)
@RabbitHandler
public void processToPre(String msg) throws InterruptedException, IOException {
Thread.sleep(500);
for (WebSocketServer webSocketServer : WebSocketServer.webSockets) {
System.out.println("WebSocket從隊列中取出客戶端（"+webSocketServer.toString()+"）發送過來的消息：");
webSocketServer.onMessage(msg);
}
}
}
再寫一個發送消息的接口，方便測試用
package com.example.demo.controller;
import com.example.demo.direct.DirectSender;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.RestController;
/**
* @author lyd
* @Description:
* @date 14:49
*/
@RestController
public class MessageController {
@Autowired
DirectSender directSender;
/**
* 接口：
* 調用向隊列中發送消息的方法
*/
@RequestMapping("senderMsg")
@ResponseBody
public void senderMsg() {
directSender.sendDirect("我是向隊列中存儲的消息");
}
}
最后寫一個html頁面，用於前端跟客戶端建立連接
<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<title>Insert title here</title>
</head>
<body>
</body>
<script type="text/javascript">
var websocket = null;
if ('WebSocket' in window) {
websocket = new WebSocket('ws://localhost:8080/webSocket');
} else {
alert('該瀏覽器不支持websocket');
}
websocket.onopen = function (e) {
console.log('websocket建立連接');
websocket.send('websocket建立連接');
}
websocket.onclose = function (e) {
console.log('websocket關閉連接');
}
websocket.onmessage = function (e) {
console.log(e, 'websocket收到消息');
document.getElementById('msgs').innerHTML = document.getElementById('msgs').innerHTML + '<br/>' + e.data;
}
websocket.onerror = function (event) {
console.log('websocket通信發生錯誤');
}
window.onbeforeunload = function (event) {
websocket.close();
}
</script>
</html>
測試
1）啟動項目后訪問http://localhost:8080/chat.html,客戶端跟服務端建立連接
2） 調用http://localhost:8080/senderMsg，WebSocket向RabbitMQ隊列中發送消息，這一步也同時實現了服務端監聽隊列消息並把消息發給WebSocket
