pub/sub

发布订阅模式

核心点

  • Pub/Sub 模型定义了如何向一个内容节点发布和订阅消息，这些节点被称作主题(topic)。
  • 主题可以被认为是消息的传输中介，发布者(publisher)发布消息到主题，订阅者(subscriber) 从主题订阅消息。
  • 主题使得消息订阅者和消息发布者保持互相独立，不需要接触即可保证消息的传送。

pub/sub解决了什么样的问题？

  • 耗时的问题，比如上传，格式转换、计算等其他耗时任务，消耗的时间大于http服务器的超时时间
  • 类似于im，一对一，一对多、群组等

技术选型

faye，最早用的时候是写ruby的时候，那时候为了用faye而了解node，因为node版本的faye效率比ruby的高非常多，也算是“因ruby得node”吧。

Simple pub/sub messaging for the web http://faye.jcoglan.com

集成

安装

  1. $ npm i -S faye

最简单例子

  1. var http = require('http'),
  2.     faye = require('faye');
  4. var server = http.createServer(),
  5.     bayeux = new faye.NodeAdapter({mount: '/'});
  7. bayeux.attach(server);
  8. server.listen(8000);

很明显，它是attach到http的server实例上的，所以我们也需要把koa改成对应的

使用类似于express的www写法

  1. #!/usr/bin/env node
  3. var faye = require('faye'); 
  5. var bayeux = new faye.NodeAdapter({
  6.     mount        : '/faye', 
  7.     timeout : 45
  8. });
  10. /**
  11.  * Module dependencies.
  12.  */
  14. var app = require('./app');
  15. // console.dir(app)
  17. var debug = require('debug')('demo:server');
  18. var http = require('http');
  20. /**
  21.  * Get port from environment and store in Express.
  22.  */
  24. var port = normalizePort(process.env.PORT || '3000');
  25. // app.set('port', port);
  27. /**
  28.  * Create HTTP server.
  29.  */
  31. var server = http.createServer(app.callback());
  33. bayeux.attach(server);
  35. /**
  36.  * Listen on provided port, on all network interfaces.
  37.  */
  39. server.listen(port);
  40. server.on('error', onError);
  41. server.on('listening', onListening);
  43. /**
  44.  * Normalize a port into a number, string, or false.
  45.  */
  47. function normalizePort(val) {
  48.   var port = parseInt(val, 10);
  50.   if (isNaN(port)) {
  51.     // named pipe
  52.     return val;
  53.   }
  55.   if (port >= 0) {
  56.     // port number
  57.     return port;
  58.   }
  60.   return false;
  61. }
  63. /**
  64.  * Event listener for HTTP server "error" event.
  65.  */
  67. function onError(error) {
  68.   if (error.syscall !== 'listen') {
  69.     throw error;
  70.   }
  72.   var bind = typeof port === 'string'
  73.     ? 'Pipe ' + port
  74.     : 'Port ' + port;
  76.   // handle specific listen errors with friendly messages
  77.   switch (error.code) {
  78.     case 'EACCES':
  79.       console.error(bind + ' requires elevated privileges');
  80.       process.exit(1);
  81.       break;
  82.     case 'EADDRINUSE':
  83.       console.error(bind + ' is already in use');
  84.       process.exit(1);
  85.       break;
  86.     default:
  87.       throw error;
  88.   }
  89. }
  91. /**
  92.  * Event listener for HTTP server "listening" event.
  93.  */
  95. function onListening() {
  96.   var addr = server.address();
  97.   var bind = typeof addr === 'string'
  98.     ? 'pipe ' + addr
  99.     : 'port ' + addr.port;
  100.   debug('Listening on ' + bind);
  101. }

其实核心很简单

  1. var server = http.createServer(app.callback());

这是koa和http的集成，所以有了server实例就可以加faye了。

  1. bayeux.attach(server);

至此，一个建的服务就构建好。

模式1：客户端订阅，服务器端发布

构建前端

  1. extends layouts/layout
  3. block content
  4.   h1= title
  5.   p Welcome to StuQ Koa小班课！
  6.   div
  7.     a(href='/users/register') 注册 
  8.     span |
  9.     a(href='/users/login') 登录 
  10.     span |
  11.     a(href='/users/logout') 销毁
  12.   script(src="http://127.0.0.1:3000/faye/client.js")
  13.   script.
  14.     var client = new Faye.Client('http://127.0.0.1:3000/faye', {
  15.       timeout : 120,
  16.           retry       : 5
  17.       });
  19.       Logger = {
  20.         incoming: function(message, callback) {
  21.           console.log('incoming', message);
  22.           callback(message);
  23.         },
  24.         outgoing: function(message, callback) {
  25.           console.log('outgoing', message);
  26.           callback(message);
  27.         }
  28.       };
  30.       client.addExtension(Logger);
  32.       client.on('transport:down', function() {
  33.         // the client is offline
  34.       });
  36.       client.on('transport:up', function() {
  37.         // the client is online
  38.       });
  40.       var subscription = client.subscribe('/messages', function(message) {
  41.         // handle message
  42.           alert(message.text);
  43.       });

服务器端发布

  1. "use strict"
  3. var Faye = require('faye'); 
  5. const router = require('koa-router')()
  8. var client = new Faye.Client('http://127.0.0.1:3000/faye');
  10. console.log(client)
  12. // 首页
  13. router.get('/', ctx => {
  14.   var session = ctx.session;
  16.   setTimeout(function(){
  17.     console.log('publish')
  18.     client.publish('/messages', {
  19.       text: 'Hello world'
  20.     });
  21.   },3000)
  24.   if (session.current_user) {
  25.     return ctx.render('index', {
  26.       title: '演示session用法: 当前状态为已登录或注册成功'
  27.     });
  28.   } else {
  29.     return ctx.render('index', {
  30.       title: '演示session用法: 当前状态为游客'
  31.     });
  32.   }
  33. });
  35. module.exports = router

核心

  1.     client.publish('/messages', {
  2.       text: 'Hello world'
  3.     });

模式2：客户端订阅，客户端发布

  1.     <script type="text/javascript">
  2.       var client = new Faye.Client('http://127.0.0.1:3000/faye', {
  3.         timeout : 120,
  4.         retry       : 5
  5.     });
  7.     Logger = {
  8.       incoming: function(message, callback) {
  9.         console.log('incoming', message);
  10.         callback(message);
  11.       },
  12.       outgoing: function(message, callback) {
  13.         console.log('outgoing', message);
  14.         callback(message);
  15.       }
  16.     };
  18.     client.addExtension(Logger);
  20.     client.on('transport:down', function() {
  21.       // the client is offline
  22.     });
  24.     client.on('transport:up', function() {
  25.       // the client is online
  26.     });
  29.     var subscription = client.subscribe('/foo', function(message) {
  30.       // handle message
  31.         console.log(message);
  32.     });
  34.     setTimeout(function(){
  35.         var publication = client.publish('/foo', {text: 'Hi there, foo test'});
  37.         publication.then(function() {
  38.           alert('Message received by server!');
  39.         }, function(error) {
  40.           alert('There was a problem: ' + error.message);
  41.         });
  43.     },2000);
  44. </script>

核心

  1. var publication = client.publish('/foo', {text: 'Hi there, foo test'});

是不是有点像im？

效率问题

它有engine的概念，即存储引擎

https://faye.jcoglan.com/node/engines.html

  • faye-redis
  • faye-redis-sharded
  • faye-couchbase 
  1. var faye      = require('faye'),
  2.     fayeRedis = require('faye-redis');
  4. var server = new faye.NodeAdapter({
  5.   mount:    '/faye',
  6.   timeout:  45,
  7.   engine:   {
  8.     type:   fayeRedis,
  9.     host:   'localhost',
  10.     port:   6379
  11.   }
  12. });

服务化

把这样的pub/sub丢出去，变成http接口，会不会更简单易用

https://github.com/i5ting/fayeserver