一涯捻、前言
websocket
和socket.io
區(qū)別浅妆?
websocket
- 一種讓客戶端和服務(wù)器之間能進(jìn)行雙向?qū)崟r(shí)通信的技術(shù)
- 使用時(shí),雖然主流瀏覽器都已經(jīng)支持障癌,但仍然可能有不兼容的情況
- 適合用于client和基于node搭建的服務(wù)端使用
socket.io
- 將WebSocket凌外、AJAX和其它的通信方式全部封裝成了統(tǒng)一的通信接口
- 使用時(shí),不用擔(dān)心兼容問(wèn)題涛浙,底層會(huì)自動(dòng)選用最佳的通信方式
- 適合進(jìn)行服務(wù)端和客戶端雙向數(shù)據(jù)通信
w3cschool上對(duì)socket.io的描述如下:
本文將實(shí)現(xiàn)
- 基于
springboot2.1.8.RELEASE
集成netty-socketio
: 仿node.js
實(shí)現(xiàn)的socket.io服務(wù)端
- 集成
socket.io-client
:socket.io客戶端
- 實(shí)現(xiàn)
服務(wù)端
與客戶端
之間的通信
二康辑、Java
集成socket.io
服務(wù)端
1摄欲、pom.xml
中引入所需依賴
溫馨小提示:這里為了方便將后面需要的客戶端
socket.io-client
依賴一起直接引入了哦~
<!-- netty-socketio: 仿`node.js`實(shí)現(xiàn)的socket.io服務(wù)端 -->
<dependency>
<groupId>com.corundumstudio.socketio</groupId>
<artifactId>netty-socketio</artifactId>
<version>1.7.7</version>
</dependency>
<!-- socket.io客戶端 -->
<dependency>
<groupId>io.socket</groupId>
<artifactId>socket.io-client</artifactId>
<version>1.0.0</version>
</dependency>
2、application.yml
中配置socket.io
服務(wù)端
# netty-socketio 配置
socketio:
host: 127.0.0.1
port: 8888
# 設(shè)置最大每幀處理數(shù)據(jù)的長(zhǎng)度疮薇,防止他人利用大數(shù)據(jù)來(lái)攻擊服務(wù)器
maxFramePayloadLength: 1048576
# 設(shè)置http交互最大內(nèi)容長(zhǎng)度
maxHttpContentLength: 1048576
# socket連接數(shù)大行厍健(如只監(jiān)聽(tīng)一個(gè)端口boss線程組為1即可)
bossCount: 1
workCount: 100
allowCustomRequests: true
# 協(xié)議升級(jí)超時(shí)時(shí)間(毫秒),默認(rèn)10秒按咒。HTTP握手升級(jí)為ws協(xié)議超時(shí)時(shí)間
upgradeTimeout: 1000000
# Ping消息超時(shí)時(shí)間(毫秒)迟隅,默認(rèn)60秒,這個(gè)時(shí)間間隔內(nèi)沒(méi)有接收到心跳消息就會(huì)發(fā)送超時(shí)事件
pingTimeout: 6000000
# Ping消息間隔(毫秒)胖齐,默認(rèn)25秒玻淑∷悦埃客戶端向服務(wù)器發(fā)送一條心跳消息間隔
pingInterval: 25000
3呀伙、socket.io服務(wù)端
配置類
@Configuration
public class SocketIOConfig {
@Value("${socketio.host}")
private String host;
@Value("${socketio.port}")
private Integer port;
@Value("${socketio.bossCount}")
private int bossCount;
@Value("${socketio.workCount}")
private int workCount;
@Value("${socketio.allowCustomRequests}")
private boolean allowCustomRequests;
@Value("${socketio.upgradeTimeout}")
private int upgradeTimeout;
@Value("${socketio.pingTimeout}")
private int pingTimeout;
@Value("${socketio.pingInterval}")
private int pingInterval;
@Bean
public SocketIOServer socketIOServer() {
SocketConfig socketConfig = new SocketConfig();
socketConfig.setTcpNoDelay(true);
socketConfig.setSoLinger(0);
com.corundumstudio.socketio.Configuration config = new com.corundumstudio.socketio.Configuration();
config.setSocketConfig(socketConfig);
config.setHostname(host);
config.setPort(port);
config.setBossThreads(bossCount);
config.setWorkerThreads(workCount);
config.setAllowCustomRequests(allowCustomRequests);
config.setUpgradeTimeout(upgradeTimeout);
config.setPingTimeout(pingTimeout);
config.setPingInterval(pingInterval);
return new SocketIOServer(config);
}
}
4、socket.io服務(wù)端
服務(wù)層
服務(wù)類
public interface ISocketIOService {
/**
* 啟動(dòng)服務(wù)
*/
void start();
/**
* 停止服務(wù)
*/
void stop();
/**
* 推送信息給指定客戶端
*
* @param userId: 客戶端唯一標(biāo)識(shí)
* @param msgContent: 消息內(nèi)容
*/
void pushMessageToUser(String userId, String msgContent);
}
服務(wù)實(shí)現(xiàn)類:
@Slf4j
@Service(value = "socketIOService")
public class SocketIOServiceImpl implements ISocketIOService {
/**
* 存放已連接的客戶端
*/
private static Map<String, SocketIOClient> clientMap = new ConcurrentHashMap<>();
/**
* 自定義事件`push_data_event`,用于服務(wù)端與客戶端通信
*/
private static final String PUSH_DATA_EVENT = "push_data_event";
@Autowired
private SocketIOServer socketIOServer;
/**
* Spring IoC容器創(chuàng)建之后添坊,在加載SocketIOServiceImpl Bean之后啟動(dòng)
*/
@PostConstruct
private void autoStartup() {
start();
}
/**
* Spring IoC容器在銷(xiāo)毀SocketIOServiceImpl Bean之前關(guān)閉,避免重啟項(xiàng)目服務(wù)端口占用問(wèn)題
*/
@PreDestroy
private void autoStop() {
stop();
}
@Override
public void start() {
// 監(jiān)聽(tīng)客戶端連接
socketIOServer.addConnectListener(client -> {
log.debug("************ 客戶端: " + getIpByClient(client) + " 已連接 ************");
// 自定義事件`connected` -> 與客戶端通信 (也可以使用內(nèi)置事件剿另,如:Socket.EVENT_CONNECT)
client.sendEvent("connected", "你成功連接上了哦...");
String userId = getParamsByClient(client);
if (userId != null) {
clientMap.put(userId, client);
}
});
// 監(jiān)聽(tīng)客戶端斷開(kāi)連接
socketIOServer.addDisconnectListener(client -> {
String clientIp = getIpByClient(client);
log.debug(clientIp + " *********************** " + "客戶端已斷開(kāi)連接");
String userId = getParamsByClient(client);
if (userId != null) {
clientMap.remove(userId);
client.disconnect();
}
});
// 自定義事件`client_info_event` -> 監(jiān)聽(tīng)客戶端消息
socketIOServer.addEventListener(PUSH_DATA_EVENT, String.class, (client, data, ackSender) -> {
// 客戶端推送`client_info_event`事件時(shí),onData接受數(shù)據(jù)贬蛙,這里是string類型的json數(shù)據(jù)雨女,還可以為Byte[],object其他類型
String clientIp = getIpByClient(client);
log.debug(clientIp + " ************ 客戶端:" + data);
});
// 啟動(dòng)服務(wù)
socketIOServer.start();
// broadcast: 默認(rèn)是向所有的socket連接進(jìn)行廣播,但是不包括發(fā)送者自身阳准,如果自己也打算接收消息的話氛堕,需要給自己?jiǎn)为?dú)發(fā)送。
new Thread(() -> {
int i = 0;
while (true) {
try {
// 每3秒發(fā)送一次廣播消息
Thread.sleep(3000);
socketIOServer.getBroadcastOperations().sendEvent("myBroadcast", "廣播消息 " + DateUtil.now());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
}
@Override
public void stop() {
if (socketIOServer != null) {
socketIOServer.stop();
socketIOServer = null;
}
}
@Override
public void pushMessageToUser(String userId, String msgContent) {
SocketIOClient client = clientMap.get(userId);
if (client != null) {
client.sendEvent(PUSH_DATA_EVENT, msgContent);
}
}
/**
* 獲取客戶端url中的userId參數(shù)(這里根據(jù)個(gè)人需求和客戶端對(duì)應(yīng)修改即可)
*
* @param client: 客戶端
* @return: java.lang.String
*/
private String getParamsByClient(SocketIOClient client) {
// 獲取客戶端url參數(shù)(這里的userId是唯一標(biāo)識(shí))
Map<String, List<String>> params = client.getHandshakeData().getUrlParams();
List<String> userIdList = params.get("userId");
if (!CollectionUtils.isEmpty(userIdList)) {
return userIdList.get(0);
}
return null;
}
/**
* 獲取連接的客戶端ip地址
*
* @param client: 客戶端
* @return: java.lang.String
*/
private String getIpByClient(SocketIOClient client) {
String sa = client.getRemoteAddress().toString();
String clientIp = sa.substring(1, sa.indexOf(":"));
return clientIp;
}
}
三野蝇、Java
開(kāi)發(fā)socket.io
客戶端
-
socket.emit
:發(fā)送數(shù)據(jù)到服務(wù)端事件 -
socket.on
: 監(jiān)聽(tīng)服務(wù)端事件
@Slf4j
public class SocketIOClientLaunch {
public static void main(String[] args) {
// 服務(wù)端socket.io連接通信地址
String url = "http://127.0.0.1:8888";
try {
IO.Options options = new IO.Options();
options.transports = new String[]{"websocket"};
options.reconnectionAttempts = 2;
// 失敗重連的時(shí)間間隔
options.reconnectionDelay = 1000;
// 連接超時(shí)時(shí)間(ms)
options.timeout = 500;
// userId: 唯一標(biāo)識(shí) 傳給服務(wù)端存儲(chǔ)
final Socket socket = IO.socket(url + "?userId=1", options);
socket.on(Socket.EVENT_CONNECT, args1 -> socket.send("hello..."));
// 自定義事件`connected` -> 接收服務(wù)端成功連接消息
socket.on("connected", objects -> log.debug("服務(wù)端:" + objects[0].toString()));
// 自定義事件`push_data_event` -> 接收服務(wù)端消息
socket.on("push_data_event", objects -> log.debug("服務(wù)端:" + objects[0].toString()));
// 自定義事件`myBroadcast` -> 接收服務(wù)端廣播消息
socket.on("myBroadcast", objects -> log.debug("服務(wù)端:" + objects[0].toString()));
socket.connect();
while (true) {
Thread.sleep(3000);
// 自定義事件`push_data_event` -> 向服務(wù)端發(fā)送消息
socket.emit("push_data_event", "發(fā)送數(shù)據(jù) " + DateUtil.now());
}
} catch (Exception e) {
e.printStackTrace();
}
}
}
四讼稚、運(yùn)行測(cè)試
當(dāng)客戶端上線后,會(huì)通過(guò)自定義事件push_data_event
每隔3秒向服務(wù)端發(fā)送消息绕沈,日志如下
而服務(wù)端中跑了一個(gè)廣播消息(自定義事件myBroadcast
) 每隔3秒也會(huì)返回給客戶端
廣播事件
: 向所有的socket連接進(jìn)行廣播發(fā)送消息數(shù)據(jù)
socketIOServer.getBroadcastOperations().sendEvent("myBroadcast", "廣播消息 " + DateUtil.now());
日志如下:
編寫(xiě)服務(wù)端主動(dòng)發(fā)送消息給客戶端接口
@RestController
@RequestMapping("/api/socket.io")
@Api(tags = "SocketIO測(cè)試-接口")
public class SocketIOController {
@Autowired
private ISocketIOService socketIOService;
@PostMapping(value = "/pushMessageToUser", produces = Constants.CONTENT_TYPE)
@ApiOperation(value = "推送信息給指定客戶端", httpMethod = "POST", response = ApiResult.class)
public ApiResult pushMessageToUser(@RequestParam String userId, @RequestParam String msgContent) {
socketIOService.pushMessageToUser(userId, msgContent);
return ApiResult.ok();
}
}
調(diào)用接口測(cè)試發(fā)送helloworld...
五锐想、總結(jié)
socket.io通信,服務(wù)端:
1.socketIOServer.addConnectListener
:監(jiān)聽(tīng)客戶端連接
-
socketIOServer.addDisconnectListener
:監(jiān)聽(tīng)客戶端斷開(kāi)連接 -
socketIOServer.addEventListener
:監(jiān)聽(tīng)客戶端傳輸?shù)南?/li> -
client.sendEvent("自定義事件名稱", "消息內(nèi)容")
:服務(wù)端向指定的clien客戶端發(fā)送消息 -
socketIOServer.getBroadcastOperations().sendEvent("自定義事件名稱", "消息內(nèi)容")
:服務(wù)端發(fā)送廣播消息給所有客戶端
socket.io通信乍狐,客戶端:
-
IO.socket(url)
:與指定的socket.io服務(wù)端建立連接 -
socket.emit
:發(fā)送數(shù)據(jù)到服務(wù)端事件 -
socket.on
: 監(jiān)聽(tīng)服務(wù)端事件