溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
socket.io與cluster
在線上系統中，需要使用node的多進程模型，我們可以自己實現簡易的基于cluster模式的socket分發模型，也可以使用比較穩定的pm2這樣進程管理工具。在常規的http服務中，這套模式一切正常，可是一旦server中集成了socket.io服務就會導致ws通道建立失敗，即使通過backup的polling方式仍會出現時斷時連的現象，因此我們需要解決這種問題，讓socket.io充分利用多核。
在這里之所以提到socket.io而未說websocket服務，是因為socket.io在封裝websocket基礎上又保證了可用性。在客戶端未提供websocket功能的基礎上使用xhr polling、jsonp或forever iframe的方式進行兼容，同時在建立ws連接前往往通過幾次http輪訓確保ws服務可用，因此socket.io并不等于websocket。再往底層深入研究，socket.io其實并沒有做真正的websocket兼容，而是提供了上層的接口以及namespace服務，真正的邏輯則是在“engine.io”模塊。該模塊實現握手的http代理、連接升級、心跳、傳輸方式等，因此研究engine.io模塊才能清楚的了解socket.io實現機制。
場景重現
服務端采用express+socket.io的組合方案，搭配pm2的cluster模式，實現一個簡易的b/s通信demo：
app.js
var path = require('path');
var app = require('express')(),
server = require('http').createServer(app),
io = require('socket.io')(server);
io
.on('connection', function(socket) {
socket.on('disconnect', function() {
console.log('/: disconnect-------->')
});
socket.on('b:message', function() {
socket.emit('s:message', '/: '+port);
console.log('/: '+port)
});
});
io.of('/ws')
.on('connection', function(socket) {
socket.on('disconnect', function() {
console.log('/ws: disconnect-------->')
});
socket.on('b:message', function() {
socket.emit('/ws: message', port);
});
});
app.get('/page',function(req,res){
res.sendFile(path.join(process.cwd(),'./index.html'));
});
server.listen(8080);
index.html
<script>
var btn = document.getElementById('btn1');
btn.addEventListener('click',function(){
var socket = io.connect('http://127.0.0.1:8080/ws',{
reconnection: false
});
socket.on('connect',function(){
// 發起“腳手架安裝”請求
socket.emit('b:message',{});
socket.on('s:message',function(d){
console.log(d);
});
});
socket.on('error',function(err){
console.log(err);
})
});
</script>
pm2.json
{
"apps": [
{
"name": "ws",
"script": "./app.js",
"env": {
"NODE_ENV": "development"
},
"env_production": {
"NODE_ENV": "production"
},
"instances": 4,
"exec_mode": "cluster",
"max_restarts" : 3,
"restart_delay" : 5000,
"log_date_format" : "YYYY-MM-DD HH:mm Z",
"combine_logs" : true
}
]
}
這樣，執行命令pm2 start pm2.json即可開啟服務，訪問127.0.0.1:8080/page，點擊按鈕發起ws連接，觀察控制臺即可。
下圖清晰顯示了socket.io握手的錯誤：
可見在websocket連接建立之前多出了3個xhr請求，而websocket連接建立失敗后又多出了幾個xhr請求，同時最后兩個xhr請求失敗了。
socket.io沒有采用直接建立websocket連接的粗暴方式，而是首先通過http請求（xhr）訪問服務端的相關輪訓配置信息以及sid。此處sid類似sessionID，但是它唯一標識連接，可理解為socketId，以后每次http請求cookie中都必須攜帶sid（httponly）；
第二、三個請求用于確認連接，在socket.io中，post請求是客戶端發送消息給服務端的唯一形式，而且post響應一定是“ok”，它的“content-length”一定為2；而get請求主要用于輪訓，同時獲取服務端的相關消息，這會在下文中有體現；
第四個websocket連接請求失敗，這主要是由于與后端http握手失敗造成的；
第五個請求為xhr方式的post請求，它是作為websocket通道建立失敗后的一種兼容性處理，上文講述了socket.io的post請求只在客戶端需要發送消息給服務端時才會使用，因此，為了證實我們查看消息體：
可見，它攜帶了客戶端發出的消息類型b:message,同時包含消息體{}空對象。對應的，服務端返回“OK”；
第六個請求為xhr方式的get請求，用來獲取服務端對第五個請求的響應。
至此，大致分析了socket.io建立連接的大致過程以及連接建立失敗后如何兜底的方案，下面分析為何出現握手失敗的問題。
原因何在
實例中pm2主進程開啟了4個工作進程，由主進程偵聽8080端口并分發請求給工作進程。pm2進程在分發請求的階段采用了某種算法的均衡，如round-robin或者其他hash方式（但不是iphash），因此在socket.io客戶端連接建立階段發送的多個xhr請求，會被pm2定位到不同的worker進程中。前文中提到每個xhr請求都會攜帶sid字段標識當前連接，因此當一個攜帶sid字段的請求被pm2定位到另一個與該連接無關的worker時，就會造成請求失敗，返回{"code":1,"message":"Session ID unknown"}錯誤；即使前三次xhr握手成功，進入websocket連接升級階段，負責偵聽update事件的worker也往往不是之前的那個worder，因此導致websocket連接建立失敗。
一言以蔽之，客戶端多次請求的服務端進程不是同一個進程才導致的ws連接無法成功建立。那么如何才能解決呢？最簡單的方案就是確保客戶端的每次請求都可以定位到同一個服務進程即可。當然，分布式session同樣可以解決問題，依托第三方緩存類似redis并配合一致性hash算法，確保所有服務進程都可以獲取到連接信息，相互配合完成連接建立。但這也僅僅是作者在理論上分析的一種實現方式，并沒有測試通過，因為這種分布式架構不僅實現繁雜而且引入了相關依賴redis，不太可取。
那么下文主要針對確保客戶端的每次請求都可以定位到同一個服務進程這一點實現解決方案。
多種實現
官方實現
官方提供了一種比較輕便的架構：nginx反向代理+iphash
我們的示例demo中的http服務器只偵聽8080端口，因此必須由pm2分發請求，否則會出現端口占用的錯誤發生。但是，官方的解決方案是每個進程的socket.io服務器創建不同端口的http服務器，專注用于http握手和升級，由nginx做握手請求的代理。而且針對nginx必須設置iphash，保證同一個客戶端的多次請求定位到后端同一個服務進程。
這樣，示例demo中會占用5個端口，其中8080端口為公用的http服務器使用，其他四個端口則只用于ws連接握手。但是這四個端口卻如何選取呢？為了保證擴展性以及順序性，采用與pm2相兼容的方案。pm2會為每個worker進程分配一個id，并且將該id綁定到進程的環境變量中，那么我們就可以利用該worker id生成4個不同的端口號。
app.js
var path = require('path');
var app = require('express')(),
server = require('http').createServer(app),
port = 3131 + parseInt(process.env.NODE_APP_INSTANCE),
io = require('socket.io')(port);
io
.on('connection', function(socket) {
socket.on('disconnect', function() {
console.log('/: disconnect-------->')
});
socket.on('b:message', function() {
socket.emit('s:message', '/: '+port);
console.log('/: '+port)
});
});
io.of('/ws')
.on('connection', function(socket) {
socket.on('disconnect', function() {
console.log('disconnect-------->')
});
socket.on('b:message', function() {
socket.emit('s:message', port);
});
});
app.get('/abc',function(req,res){
res.sendFile(path.join(process.cwd(),'./index.html'));
});
server.listen(8080);
index.html
<script>
var btn = document.getElementById('btn1');
btn.addEventListener('click',function(){
var socket = io.connect('http://ws.vd.net/ws',{
reconnection: false
});
socket.on('connect',function(){
// 發起“腳手架安裝”請求
socket.emit('b:message',{a:1});
socket.on('s:message',function(d){
console.log(d);
});
});
socket.on('error',function(err){
console.log(err);
})
});
</script>
nginx.conf
upstream io_nodes {
ip_hash;
server 127.0.0.1:3131;
server 127.0.0.1:3132;
server 127.0.0.1:3133;
server 127.0.0.1:3134;
}
server {
listen 80;
server_name ws.vd.net;
location / {
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header Host $host;
proxy_http_version 1.1;
proxy_pass http://io_nodes;
}
}
在本機綁定hosts地址后開啟nginx服務，同時開啟服務器，點擊按鈕建立ws連接成功。
服務端路由
服務端路由，意義在于“服務端做worker的負載均衡，并將選擇的worker ip和端口渲染在頁面，之后瀏覽器的所有ws連接默認連接到對應 ip:port的服務器中”。這樣只要是服務端渲染的頁面都可以采用這種方式實現。
如果頁面采用前端異步渲染，仍可以采用這種方式，不過首先通過xhr請求向服務端獲取需要握手的http服務器的ip和端口，然后在進行ws連接。
服務端路由的前提仍然是需要針對每個ws服務器分配一個端口，只不過去掉nginx由服務端做ip hash。采用服務端路由架構清晰，而且實現容易，兼容性好。
上帝進程路由
此處的上帝進程即為主進程，類似pm2進程。上帝進程路由則是在上帝進程層面上做請求的定向分發，保證請求主機和進程的一致性。在上帝進程中，針對每個請求的ip做hash，并對每一個ws服務器創建單獨的http服務器用于握手升級。
簡易代碼:
var express = require('express'),
cluster = require('cluster'),
net = require('net'),
sio = require('socket.io');
var port = 3000,
num_processes = require('os').cpus().length;
if (cluster.isMaster) {
var workers = [];
var spawn = function(i) {
workers[i] = cluster.fork();
workers[i].on('exit', function(code, signal) {
console.log('respawning worker', i);
spawn(i);
});
};
for (var i = 0; i < num_processes; i++) {
spawn(i);
}
// ip hash
var worker_index = function(ip, len) {
var s = '';
for (var i = 0, _len = ip.length; i < _len; i++) {
if (!isNaN(ip[i])) {
s += ip[i];
}
}
return Number(s) % len;
};
var server = net.createServer({ pauseOnConnect: true }, function(connection) {
var worker = workers[worker_index(connection.remoteAddress, num_processes)];
worker.send('sticky-session:connection', connection);
}).listen(port);
} else {
// worker
var app = new express();
// handshake server.
var server = app.listen(0, 'localhost'),
io = sio(server);
process.on('message', function(message, connection) {
if (message !== 'sticky-session:connection') {
return;
}
server.emit('connection', connection);
connection.resume();
});
}
總結
本文實現了三種解決方案，歸根到底就是“ip hash”，不同點在于在請求處理的不同階段做ip hash。
可以在請求處理最前端做iphash，即nginx方式，這也就是第一種方案；
可以在請求處理的第二層分發處做iphash，即上帝進程路由的方式，即第三種；
也可以在請求處理的終端做iphash，即服務端路由的方式，也就是第二種；
同時共享session也同樣可以實現，借助socket.io-redis模塊也可以實現。
好了，以上就是這篇文章的全部內容了，希望本文的內容對大家的學習或者工作能帶來一定的幫助，如果有疑問大家可以留言交流，謝謝大家對億速云的支持。
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
