From c1d70a5035640d6679d3354734167239690526c2 Mon Sep 17 00:00:00 2001 From: xiaozhong Date: Thu, 19 Apr 2018 17:18:47 +0800 Subject: [PATCH 1/2] =?UTF-8?q?=E6=96=B0=E5=A2=9Edubbo=E5=8D=8F=E8=AE=AE?= =?UTF-8?q?=E6=96=B9=E5=BC=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- index.js | 3 +- lib/config/index.js | 10 +- lib/registry/index.js | 10 +- lib/rpc/api/invoker/desc.js | 3 + lib/rpc/api/invoker/index.js | 20 ++- lib/rpc/util/Socket.js | 233 +++++++++++++++++++++++++++++++++++ package.json | 3 +- test/dubbo.config.js | 13 +- test/test.js | 29 +++-- 9 files changed, 306 insertions(+), 18 deletions(-) create mode 100644 lib/rpc/util/Socket.js diff --git a/index.js b/index.js index 36b7bf2..3b4c05f 100644 --- a/index.js +++ b/index.js @@ -22,7 +22,8 @@ module.exports = { }, /** - * rpc调用, 主要调用入口 + * jsonrpc || dubbo 调用, 主要调用入口 + * 不能同时使用jsonrpc或dubbo协议调用服务 */ getService: function (serviceName, version, group) { var invokerDesc = new InvokerDesc(serviceName, group, version), diff --git a/lib/config/index.js b/lib/config/index.js index 58b0e5e..a5b6c04 100644 --- a/lib/config/index.js +++ b/lib/config/index.js @@ -30,7 +30,8 @@ Config.prototype = { this.option = option; this.option.dubbo = this.option.dubbo || { providerTimeout: 3, - weight: 1 + weight: 1, + protocol: "jsonrpc" //添加协议类型，支持rpc调用 }; }, @@ -73,6 +74,13 @@ Config.prototype = { */ getProviderTimeout: function(){ return this.option.dubbo.providerTimeout * 1000; + }, + + /** + * 获取protocol协议 + */ + getProtocol: function(){ + return this.option.dubbo.protocol; } }; diff --git a/lib/registry/index.js b/lib/registry/index.js index b744202..e8a177d 100644 --- a/lib/registry/index.js +++ b/lib/registry/index.js @@ -24,7 +24,7 @@ Registry.prototype.init = function () { this.isInitializing = true; this.zookeeper = Zookeeper.createClient(Config.getRegistryAddress(), Config.getRegistryOption()); this.zookeeper.once('connected', function () { - self.initQueue.forEach(function (p) { //从队列中获取, 租个通知 + self.initQueue.forEach(function (p) { //从队列中获取, 逐个通知 p.resolve(self.zookeeper); }); self.isInitializing = false; @@ -122,8 +122,12 @@ Registry.prototype.subscribe = function (invokerDesc) { Registry.prototype.onMethodChangeHandler = function (invokerDesc, children) { children.forEach(function (child) { child = decodeURIComponent(child); - var mHost = /^jsonrpc:\/\/([^\/]+)\//.exec(child), - mVersion = /version=(.+)/.exec(child), + var mHost = /^jsonrpc:\/\/([^\/]+)\//.exec(child); + if (Config.getProtocol() === "dubbo") { + mHost = /^dubbo:\/\/([^\/]+)\//.exec(child); + } + + var mVersion = /version=(.+)/.exec(child), mGroup = /group=([^&]+)/.exec(child), mMehtod = /methods=([^&]+)/.exec(child); diff --git a/lib/rpc/api/invoker/desc.js b/lib/rpc/api/invoker/desc.js index 71cff17..fbf638e 100644 --- a/lib/rpc/api/invoker/desc.js +++ b/lib/rpc/api/invoker/desc.js @@ -25,6 +25,9 @@ InvokerDesc.prototype.getService = function () { return this.serviceName; }; +InvokerDesc.prototype.getVersion = function() { + return this.version; +}; //----------------------------------------------------------------------------------------------- // //----------------------------------------------------------------------------------------------- diff --git a/lib/rpc/api/invoker/index.js b/lib/rpc/api/invoker/index.js index 5cc28cb..b11c559 100644 --- a/lib/rpc/api/invoker/index.js +++ b/lib/rpc/api/invoker/index.js @@ -1,7 +1,8 @@ var Q = require('q'), _ = require('underscore'), Cluster = require('../../cluster/index'), - HttpClient = require('../../util/Http'); + HttpClient = require('../../util/Http'), + SocketClient = require('../../util/Socket'); //----------------------------------------------------------------------------------------------- // @@ -29,7 +30,7 @@ var time = new Date().getTime(), //----------------------------------------------------------------------------------------------- -// 对外只有一个方法, 和代理提供的方法 +// 对外只有一个方法, 和代理提供的call jsonrpc方法 //----------------------------------------------------------------------------------------------- Invoker.prototype.call = function (methodName) { var desc = this.invokerDesc, @@ -44,6 +45,21 @@ Invoker.prototype.call = function (methodName) { }); }; +//----------------------------------------------------------------------------------------------- +// 对外只有一个方法, 和代理提供的callrpc方法 +//----------------------------------------------------------------------------------------------- +Invoker.prototype.callRpc = function (methodName) { + console.log("request callRpc methodName [" + methodName + "]"); + var desc = this.invokerDesc, + service = desc.serviceName, + methodArgs = _.toArray(arguments).slice(1); + + return Cluster + .getProvider(desc) + .then(function (serverMap) { + return SocketClient.execute(serverMap, service, methodName, desc.getVersion(), methodArgs) + }); +}; //----------------------------------------------------------------------------------------------- // // diff --git a/lib/rpc/util/Socket.js b/lib/rpc/util/Socket.js new file mode 100644 index 0000000..fc7e2cb --- /dev/null +++ b/lib/rpc/util/Socket.js @@ -0,0 +1,233 @@ +//引入hession.js +const hessian = require('hessian.js'); +const net = require('net'); //网络库 +const url = require('url'); +const qs = require('querystring'); +const Q = require('q'); +const DEFAULT_LEN = 8388608; // 8 * 1024 * 1024 //默认参数buffer长度 + +const ERROR = { + '100': 'create buffer failed ', + '102': '连接服务失败 ', + '103': '服务不可用', + '104': '未找到对应的method', + '105': 'socket连接错误', + '106': '远程服务无响应，请重试', + '107': '该服务未返回任何数据', + '108': '请求参数不合法或类型不一致', + '109': 'hessian decoder error ', + '110': 'socket已经关闭', +}; + +var SocketClient = function() { + this._dubboVersion = '2.5.3'; + this._group = ''; +}; + +SocketClient.prototype.execute = function(server, service, method, version, args) { + console.log("server: " + JSON.stringify(server) + " service = " + service + " method = " + method + " args = " + JSON.stringify(args)); + var q = Q.defer(); + var _this = this, + buffer; + try { + buffer = _this._createBuffer(service, method, version, args); + } catch (e) { + q.reject({ + code: '100', + error: e.message || ERROR['100'] + }); + } + + _this._fetchData(_this.serverIpAndPort(server), buffer, method) + .then(function(data) { + q.resolve(data); + }) + .catch(function(err) { + q.reject(err); + }); + return q.promise; +}; + +SocketClient.prototype.serverIpAndPort = function(server) { + var ips = server.split(":"); + return { + host: ips[0], + port: ips[1] + }; +}; + +SocketClient.prototype._fetchData = function(zoo, buffer, method) { + var q = Q.defer(); + + var bl = 16; + var host = zoo.host; + var port = zoo.port; + var ret = null; + var chunks = []; + var tryCount = 0; + var heap, ret; + + var client = new net.Socket(); + + //connect + client.connect(port, host, function() { + client.write(buffer); + }); + + //发送数据 + client.on('data', function(chunk) { + + if (!chunks.length) { + var arr = Array.prototype.slice.call(chunk.slice(0, 16)); + var i = 0; + while (i < 3) { + bl += arr.pop() * Math.pow(255, i++); + } + } + + chunks.push(chunk); + heap = Buffer.concat(chunks); + (heap.length >= bl) && client.destroy(); + }); + + // 105 socket connection error + client.on('error', function(err) { + client.destroy(); + q.reject({ + code: '105', + error: ERROR['105'] + (err.message || err) + }); + }); + + // 110 socket closed + client.on('close', function(err) { + + if (err) { + q.reject({ + code: '110', + error: ERROR['110'] + (err.message || '') + }); + } + // 106 service response error + if (heap[3] !== 20) { + ret = heap.slice(18, heap.length - 1).toString(); + q.reject({ + code: '106', + error: ERROR['106'] + ret + }); + } + + // 107 service void return + if (heap[3] === 20 && heap[15] === 1) { + q.resolve(true); + } + + try { + var offset = heap[16] === 145 ? 17 : 18; + var buf = new hessian.DecoderV2(heap.slice(offset, heap.length)); + var _ret = buf.read(); + if (_ret instanceof Error || offset === 18) { + q.reject({ + code: '108', + error: ERROR['108'] + (_ret.message || _ret) + }); //108 hessian read error + } + q.resolve(_ret); + } catch (e) { + q.reject({ + code: '109', + error: ERROR['109'] + (e.message || e) + }); //109 hessian decoder error + } + + }); + + client.on('timeout', function() { + client.destroy(); + console.log('socket timeout'); + }); + + return q.promise; +}; + +SocketClient.prototype._createBuffer = function(service, method, version, args) { + var typeRef, types, type, buffer; + + typeRef = { + boolean: 'Z', + int: 'I', + short: 'S', + long: 'J', + double: 'D', + float: 'F' + }; + + if (args && args.length) { + for (var i = 0, l = args.length; i < l; i++) { + type = args[i]['$class']; + types += type && ~type.indexOf('.') ? 'L' + type.replace(/\./gi, '/') + ';' : typeRef[type]; + } + buffer = this.buffer(service, method, version, types, args); + } else { + buffer = this.buffer(method, ''); + } + + return buffer; +}; + +SocketClient.prototype.buffer = function(service, method, version, type, args) { + var bufferBody = this.bufferBody(service, method, version, type, args); + var bufferHead = this.bufferHead(bufferBody.length); + return Buffer.concat([bufferHead, bufferBody]); +}; + +SocketClient.prototype.bufferHead = function(length) { + var head = [0xda, 0xbb, 0xc2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]; + var i = 15; + + if (length > DEFAULT_LEN) { + throw new Error(`Data length too large: ${length}, max payload: ${DEFAULT_LEN}`); + } + // 构造body长度信息 + if (length - 256 < 0) { + head.splice(i, 1, length - 256); + } else { + while (length - 256 > 0) { + head.splice(i--, 1, length % 256); + length = length >> 8; + } + head.splice(i, 1, length); + } + return new Buffer(head); +}; + +SocketClient.prototype.bufferBody = function(service, method, version, type, args) { + var encoder = new hessian.EncoderV2(); + encoder.write(this._dubboVersion); + encoder.write(service); + encoder.write(version); + encoder.write(method); + encoder.write(type); + + if (args && args.length) { + for (var i = 0, len = args.length; i < len; ++i) { + encoder.write(args[i]); + } + } + + encoder.write({ + $class: 'java.util.HashMap', + $: { + interface: service, + version: version, + group: this._group, + path: service, + timeout: '60000' + } + }); + + encoder = encoder.byteBuffer._bytes.slice(0, encoder.byteBuffer._offset); + return encoder; +}; + +module.exports = new SocketClient(); diff --git a/package.json b/package.json index 4f28664..b022362 100644 --- a/package.json +++ b/package.json @@ -20,6 +20,7 @@ "node-zookeeper-client":"0.2.2", "q": "*", "underscore": "*", - "request": "*" + "request": "*", + "hessian.js": "^2.1.8" } } diff --git a/test/dubbo.config.js b/test/dubbo.config.js index 2b33b9f..174de43 100644 --- a/test/dubbo.config.js +++ b/test/dubbo.config.js @@ -17,7 +17,16 @@ module.exports = { /** * 注册中心 */ - registry: '172.19.65.33:2181', + registry: '172.16.150.60:2181', + + /** + * dubbo config + */ + dubbo: { + providerTimeout: 3, + weight: 1, + protocol: "dubbo" //添加协议类型，支持rpc调用 默认为jsonrpc, 指定协议则初始化过程中只按照配置协议加载到缓存 + }, /** * 负载均衡规则, 目前只有轮询 @@ -27,5 +36,5 @@ module.exports = { /** * 懒加载, 用于开发阶段, 快速启动 */ - lazy: true + lazy: false }; \ No newline at end of file diff --git a/test/test.js b/test/test.js index 6d98be1..5f77fc4 100644 --- a/test/test.js +++ b/test/test.js @@ -5,19 +5,32 @@ var _ = require('underscore'), dubboClient.config(require('./dubbo.config.js')); ////获取serivce -var catQueryProvider = dubboClient.getService('com.qianmi.pc.api.cat.StandardCatQueryProvider', '1.2.3'); +var catQueryProvider = dubboClient.getService('com.lefit.dubbo.coach.api.goods.GoodsSkuService', '1.0'); -//setTimeout(function(){ -// for(var k in catQueryProvider){ -// console.info(k); -// } -//}, 1000) - -catQueryProvider.call('listByParentId', 111) +// dubbo协议方式调用 +var data = { + "$class": "com.lefit.dubbo.coach.api.goods.request.LessonSkuTypeReq", + "$": {"goodsNo": "01241", "coachId": 120, "userId": 105424} +}; +catQueryProvider.callRpc('goodsSkuDetail', data) .then(function (r) { console.info(r); + process.exit(0); }) .catch(function (e) { console.error(JSON.stringify(e)); + process.exit(0); }); +//jsonrpc 方式调用 +// catQueryProvider.call('goodsSkuDetail', 111) +// .then(function (r) { +// console.info(r); +// process.exit(0); +// }) +// .catch(function (e) { +// console.error(JSON.stringify(e)); +// process.exit(0); +// }); + + From f223fe87c73c04e26b6f9dad6d8e9c6fafc4d82f Mon Sep 17 00:00:00 2001 From: xiaozhong Date: Thu, 19 Apr 2018 17:21:18 +0800 Subject: [PATCH 2/2] =?UTF-8?q?=E4=BF=AE=E6=94=B9=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 23 ++++++++++++++++++++++- test/test.js | 4 ++-- 2 files changed, 24 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 2b12981..632d05b 100644 --- a/README.md +++ b/README.md @@ -10,7 +10,7 @@ npm install dubbo-node-client # 注意 - 该项目只支持 [jsonrpc协议](https://github.com/ofpay/dubbo-rpc-jsonrpc), 不支持 dubbo协议的服务提供者 + 该项目只支持 [jsonrpc协议](https://github.com/ofpay/dubbo-rpc-jsonrpc), 支持 dubbo协议的服务提供者 --- @@ -60,4 +60,25 @@ function doFoo(req, res){ }); } + +//新增 DUBBO方式调用如下： + +////获取serivce +var GoodsSkuService = dubboClient.getService('com.xxx.dubbo.coach.api.goods.GoodsSkuService', '1.0'); + +// dubbo协议方式调用 以私教为例 +var data = { + "$class": "com.xxx.dubbo.coach.api.goods.request.LessonSkuTypeReq", + "$": {"goodsNo": "01241", "coachId": 120, "userId": 105424} +}; +GoodsSkuService.callRpc('goodsSkuDetail', data) + .then(function (r) { + console.info(r); + process.exit(0); + }) + .catch(function (e) { + console.error(JSON.stringify(e)); + process.exit(0); + }); +} ``` \ No newline at end of file diff --git a/test/test.js b/test/test.js index 5f77fc4..3898454 100644 --- a/test/test.js +++ b/test/test.js @@ -5,11 +5,11 @@ var _ = require('underscore'), dubboClient.config(require('./dubbo.config.js')); ////获取serivce -var catQueryProvider = dubboClient.getService('com.lefit.dubbo.coach.api.goods.GoodsSkuService', '1.0'); +var catQueryProvider = dubboClient.getService('com.xxx.dubbo.coach.api.goods.GoodsSkuService', '1.0'); // dubbo协议方式调用 var data = { - "$class": "com.lefit.dubbo.coach.api.goods.request.LessonSkuTypeReq", + "$class": "com.xxx.dubbo.coach.api.goods.request.LessonSkuTypeReq", "$": {"goodsNo": "01241", "coachId": 120, "userId": 105424} }; catQueryProvider.callRpc('goodsSkuDetail', data)