Socket.IO


Socket.IO 是一个基于 Node.js 的实时应用程序框架,在即时通讯、通知与消息推送,实时分析等场景中有较为广泛的应用。

WebSocket 的产生源于 Web 开发中日益增长的实时通信需求,对比基于 http 的轮询方式,它大大节省了网络带宽,同时也降低了服务器的性能消耗; socket.io 支持 websocket、polling 两种数据传输方式以兼容浏览器不支持 WebSocket 场景下的通信需求。

框架提供了 egg-socket.io 插件,增加了以下开发规约:

  • namespace: 通过配置的方式定义 namespace(命名空间)
  • middleware: 对每一次 socket 连接的建立/断开、每一次消息/数据传递进行预处理
  • controller: 响应 socket.io 的 event 事件
  • router: 统一了 socket.io 的 event 与 框架路由的处理配置方式

安装 egg-socket.io

安装

  1. $ npm i egg-socket.io --save

开启插件:

  1. // {app_root}/config/plugin.js
  2. exports.io = {
  3. enable: true,
  4. package: 'egg-socket.io',
  5. };

配置

  1. // {app_root}/config/config.${env}.js
  2. exports.io = {
  3. init: { }, // passed to engine.io
  4. namespace: {
  5. '/': {
  6. connectionMiddleware: [],
  7. packetMiddleware: [],
  8. },
  9. '/example': {
  10. connectionMiddleware: [],
  11. packetMiddleware: [],
  12. },
  13. },
  14. };

命名空间为 //example, 不是 example

uws

Egg Socket 内部默认使用 ws 引擎,uws 因为某些原因被废止。

如坚持需要使用,请按照以下配置即可:

  1. // {app_root}/config/config.${env}.js
  2. exports.io = {
  3. init: { wsEngine: 'uws' }, // default: ws
  4. };

redis

egg-socket.io 内置了 socket.io-redis,在 cluster 模式下,使用 redis 可以较为简单的实现 clients/rooms 等信息共享

  1. // {app_root}/config/config.${env}.js
  2. exports.io = {
  3. redis: {
  4. host: { redis server host },
  5. port: { redis server port },
  6. auth_pass: { redis server password },
  7. db: 0,
  8. },
  9. };

开启 redis 后,程序在启动时会尝试连接到 redis 服务器 此处 redis 仅用于存储连接实例信息,参见 #server.adapter

注意: 如果项目中同时使用了 egg-redis, 请单独配置,不可共用。

部署

框架是以 Cluster 方式启动的,而 socket.io 协议实现需要 sticky 特性支持,否则在多进程模式下无法正常工作。

由于 socket.io 的设计,在多进程中服务器必须在 sticky 模式下工作,故需要给 startCluster 传递 sticky 参数。

修改 package.jsonnpm scripts 脚本:

  1. {
  2. "scripts": {
  3. "dev": "egg-bin dev --sticky",
  4. "start": "egg-scripts start --sticky"
  5. }
  6. }

Nginx 配置

  1. location / {
  2. proxy_set_header Upgrade $http_upgrade;
  3. proxy_set_header Connection "upgrade";
  4. proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
  5. proxy_set_header Host $host;
  6. proxy_pass http://127.0.0.1:7001;
  7. # http://nginx.org/en/docs/http/ngx_http_proxy_module.html#proxy_bind
  8. # proxy_bind $remote_addr transparent;
  9. }

使用 egg-socket.io

开启 egg-socket.io 的项目目录结构如下:

  1. chat
  2. ├── app
  3. ├── extend
  4. └── helper.js
  5. ├── io
  6. ├── controller
  7. └── default.js
  8. └── middleware
  9. ├── connection.js
  10. └── packet.js
  11. └── router.js
  12. ├── config
  13. └── package.json

注意:对应的文件都在 app/io 目录下

Middleware

中间件有如下两种场景:

  • Connection
  • Packet

其配置于各个命名空间下,根据上述两种场景分别发生作用。

注意:

如果我们启用了框架中间件,则会发现项目中有以下目录:

  • app/middleware:框架中间件
  • app/io/middleware:插件中间件

区别:

  • 框架中间件基于 http 模型设计,处理 http 请求。
  • 插件中间件基于 socket 模型设计,处理 socket.io 请求。

虽然框架通过插件尽量统一了它们的风格,但务必注意,它们的使用场景是不一样的。详情参见 issue:#1416

Connection

在每一个客户端连接或者退出时发生作用,故而我们通常在这一步进行授权认证,对认证失败的客户端做出相应的处理

  1. // {app_root}/app/io/middleware/connection.js
  2. module.exports = app => {
  3. return async (ctx, next) => {
  4. ctx.socket.emit('res', 'connected!');
  5. await next();
  6. // execute when disconnect.
  7. console.log('disconnection!');
  8. };
  9. };

踢出用户示例:

  1. const tick = (id, msg) => {
  2. logger.debug('#tick', id, msg);
  3. socket.emit(id, msg);
  4. app.io.of('/').adapter.remoteDisconnect(id, true, err => {
  5. logger.error(err);
  6. });
  7. };

同时,针对当前的连接也可以简单处理:

  1. // {app_root}/app/io/middleware/connection.js
  2. module.exports = app => {
  3. return async (ctx, next) => {
  4. if (true) {
  5. ctx.socket.disconnect();
  6. return;
  7. }
  8. await next();
  9. console.log('disconnection!');
  10. };
  11. };

Packet

作用于每一个数据包(每一条消息);在生产环境中,通常用于对消息做预处理,又或者是对加密消息的解密等操作

  1. // {app_root}/app/io/middleware/packet.js
  2. module.exports = app => {
  3. return async (ctx, next) => {
  4. ctx.socket.emit('res', 'packet received!');
  5. console.log('packet:', this.packet);
  6. await next();
  7. };
  8. };

Controller

Controller 对客户端发送的 event 进行处理;由于其继承于 egg.Contoller, 拥有如下成员对象:

  • ctx
  • app
  • service
  • config
  • logger

详情参考 Controller 文档

  1. // {app_root}/app/io/controller/default.js
  2. 'use strict';
  3. const Controller = require('egg').Controller;
  4. class DefaultController extends Controller {
  5. async ping() {
  6. const { ctx, app } = this;
  7. const message = ctx.args[0];
  8. await ctx.socket.emit('res', `Hi! I've got your message: ${message}`);
  9. }
  10. }
  11. module.exports = DefaultController;
  12. // or async functions
  13. exports.ping = async function() {
  14. const message = this.args[0];
  15. await this.socket.emit('res', `Hi! I've got your message: ${message}`);
  16. };

Router

路由负责将 socket 连接的不同 events 分发到对应的 controller,框架统一了其使用方式

  1. // {app_root}/app/router.js
  2. module.exports = app => {
  3. const { router, controller, io } = app;
  4. // default
  5. router.get('/', controller.home.index);
  6. // socket.io
  7. io.of('/').route('server', io.controller.home.server);
  8. };

注意:

nsp 有如下的系统事件:

  • disconnecting doing the disconnect
  • disconnect connection has disconnected.
  • error Error occurred

Namespace/Room

Namespace (nsp)

namespace 通常意味分配到不同的接入点或者路径,如果客户端没有指定 nsp,则默认分配到 “/“ 这个默认的命名空间。

在 socket.io 中我们通过 of 来划分命名空间;鉴于 nsp 通常是预定义且相对固定的存在,框架将其进行了封装,采用配置的方式来划分不同的命名空间。

  1. // socket.io
  2. var nsp = io.of('/my-namespace');
  3. nsp.on('connection', function(socket){
  4. console.log('someone connected');
  5. });
  6. nsp.emit('hi', 'everyone!');
  7. // egg
  8. exports.io = {
  9. namespace: {
  10. '/': {
  11. connectionMiddleware: [],
  12. packetMiddleware: [],
  13. },
  14. },
  15. };

Room

room 存在于 nsp 中,通过 join/leave 方法来加入或者离开; 框架中使用方法相同;

  1. const room = 'default_room';
  2. module.exports = app => {
  3. return async (ctx, next) => {
  4. ctx.socket.join(room);
  5. ctx.app.io.of('/').to(room).emit('online', { msg: 'welcome', id: ctx.socket.id });
  6. await next();
  7. console.log('disconnection!');
  8. };
  9. };

注意: 每一个 socket 连接都会拥有一个随机且不可预测的唯一 id Socket#id,并且会自动加入到以这个 id 命名的 room 中

实例

这里我们使用 egg-socket.io 来做一个支持 p2p 聊天的小例子

client

UI 相关的内容不重复写了,通过 window.socket 调用即可

  1. // browser
  2. const log = console.log;
  3. window.onload = function() {
  4. // init
  5. const socket = io('/', {
  6. // 实际使用中可以在这里传递参数
  7. query: {
  8. room: 'demo',
  9. userId: `client_${Math.random()}`,
  10. },
  11. transports: ['websocket']
  12. });
  13. socket.on('connect', () => {
  14. const id = socket.id;
  15. log('#connect,', id, socket);
  16. // 监听自身 id 以实现 p2p 通讯
  17. socket.on(id, msg => {
  18. log('#receive,', msg);
  19. });
  20. });
  21. // 接收在线用户信息
  22. socket.on('online', msg => {
  23. log('#online,', msg);
  24. });
  25. // 系统事件
  26. socket.on('disconnect', msg => {
  27. log('#disconnect', msg);
  28. });
  29. socket.on('disconnecting', () => {
  30. log('#disconnecting');
  31. });
  32. socket.on('error', () => {
  33. log('#error');
  34. });
  35. window.socket = socket;
  36. };

微信小程序

微信小程序提供的 API 为 WebSocket ,而 socket.io 是 Websocket 的上层封装,故我们无法直接用小程序的 API 连接,可以使用类似 weapp.socket.io 的库来适配。

示例代码如下:

  1. // 小程序端示例代码
  2. const io = require('./yout_path/weapp.socket.io.js')
  3. const socket = io('http://localhost:8000')
  4. socket.on('connect', function () {
  5. console.log('connected')
  6. });
  7. socket.on('news', d => {
  8. console.log('received news: ', d)
  9. })
  10. socket.emit('news', {
  11. # 'this is a news'
  12. })

server

以下是 demo 的部分代码并解释了各个方法的作用

config

  1. // {app_root}/config/config.${env}.js
  2. exports.io = {
  3. namespace: {
  4. '/': {
  5. connectionMiddleware: [ 'auth' ],
  6. packetMiddleware: [ ], // 针对消息的处理暂时不实现
  7. },
  8. },
  9. // cluster 模式下,通过 redis 实现数据共享
  10. redis: {
  11. host: '127.0.0.1',
  12. port: 6379,
  13. },
  14. };
  15. // 可选
  16. exports.redis = {
  17. client: {
  18. port: 6379,
  19. host: '127.0.0.1',
  20. password: '',
  21. db: 0,
  22. },
  23. };

helper

框架扩展用于封装数据格式

  1. // {app_root}/app/extend/helper.js
  2. module.exports = {
  3. parseMsg(action, payload = {}, metadata = {}) {
  4. const meta = Object.assign({}, {
  5. timestamp: Date.now(),
  6. }, metadata);
  7. return {
  8. meta,
  9. data: {
  10. action,
  11. payload,
  12. },
  13. };
  14. },
  15. };

Format:

  1. {
  2. data: {
  3. action: 'exchange', // 'deny' || 'exchange' || 'broadcast'
  4. payload: {},
  5. },
  6. meta:{
  7. timestamp: 1512116201597,
  8. client: 'nNx88r1c5WuHf9XuAAAB',
  9. target: 'nNx88r1c5WuHf9XuAAAB'
  10. },
  11. }

middleware

egg-socket.io 中间件负责 socket 连接的处理

  1. // {app_root}/app/io/middleware/auth.js
  2. const PREFIX = 'room';
  3. module.exports = () => {
  4. return async (ctx, next) => {
  5. const { app, socket, logger, helper } = ctx;
  6. const id = socket.id;
  7. const nsp = app.io.of('/');
  8. const query = socket.handshake.query;
  9. // 用户信息
  10. const { room, userId } = query;
  11. const rooms = [ room ];
  12. logger.debug('#user_info', id, room, userId);
  13. const tick = (id, msg) => {
  14. logger.debug('#tick', id, msg);
  15. // 踢出用户前发送消息
  16. socket.emit(id, helper.parseMsg('deny', msg));
  17. // 调用 adapter 方法踢出用户,客户端触发 disconnect 事件
  18. nsp.adapter.remoteDisconnect(id, true, err => {
  19. logger.error(err);
  20. });
  21. };
  22. // 检查房间是否存在,不存在则踢出用户
  23. // 备注:此处 app.redis 与插件无关,可用其他存储代替
  24. const hasRoom = await app.redis.get(`${PREFIX}:${room}`);
  25. logger.debug('#has_exist', hasRoom);
  26. if (!hasRoom) {
  27. tick(id, {
  28. type: 'deleted',
  29. message: 'deleted, room has been deleted.',
  30. });
  31. return;
  32. }
  33. // 用户加入
  34. logger.debug('#join', room);
  35. socket.join(room);
  36. // 在线列表
  37. nsp.adapter.clients(rooms, (err, clients) => {
  38. logger.debug('#online_join', clients);
  39. // 更新在线用户列表
  40. nsp.to(room).emit('online', {
  41. clients,
  42. action: 'join',
  43. target: 'participator',
  44. message: `User(${id}) joined.`,
  45. });
  46. });
  47. await next();
  48. // 用户离开
  49. logger.debug('#leave', room);
  50. // 在线列表
  51. nsp.adapter.clients(rooms, (err, clients) => {
  52. logger.debug('#online_leave', clients);
  53. // 获取 client 信息
  54. // const clientsDetail = {};
  55. // clients.forEach(client => {
  56. // const _client = app.io.sockets.sockets[client];
  57. // const _query = _client.handshake.query;
  58. // clientsDetail[client] = _query;
  59. // });
  60. // 更新在线用户列表
  61. nsp.to(room).emit('online', {
  62. clients,
  63. action: 'leave',
  64. target: 'participator',
  65. message: `User(${id}) leaved.`,
  66. });
  67. });
  68. };
  69. };

controller

P2P 通信,通过 exchange 进行数据交换

  1. // {app_root}/app/io/controller/nsp.js
  2. const Controller = require('egg').Controller;
  3. class NspController extends Controller {
  4. async exchange() {
  5. const { ctx, app } = this;
  6. const nsp = app.io.of('/');
  7. const message = ctx.args[0] || {};
  8. const socket = ctx.socket;
  9. const client = socket.id;
  10. try {
  11. const { target, payload } = message;
  12. if (!target) return;
  13. const msg = ctx.helper.parseMsg('exchange', payload, { client, target });
  14. nsp.emit(target, msg);
  15. } catch (error) {
  16. app.logger.error(error);
  17. }
  18. }
  19. }
  20. module.exports = NspController;

router

  1. // {app_root}/app/router.js
  2. module.exports = app => {
  3. const { router, controller, io } = app;
  4. router.get('/', controller.home.index);
  5. // socket.io
  6. io.of('/').route('exchange', io.controller.nsp.exchange);
  7. };

开两个 tab 页面,并调出控制台:

  1. socket.emit('exchange', {
  2. target: 'Dkn3UXSu8_jHvKBmAAHW',
  3. payload: {
  4. msg : 'test',
  5. },
  6. });

Socket.IO - 图1

参考链接