nodejs微服务解决方案
前言
seneca是一个nodejs微服务工具集，它赋予系统易于连续构建和更新的能力。下面会逐一和你们一块儿了解相关技术入门以及实践。html
这里插入一段硬广。小子再进行简单整合以后撸了个vastify框架 ---- 轻量级nodejs微服务框架，有兴趣的同窗过目一下，欢迎顺手star一波，另外有疑问或者代码有毛病欢迎在博文下方留言。node
环境
- 基础环境
"node": "^10.0.0" "npm": "^6.0.0" "pm2": "^2.10.3" "rabbitmq": "^3.7.5" "consul": "^1.1.0" "mongodb": "^3.6"
- 微服务工程
"bluebird": "^3.5.1" "koa": "^2.5.1" "koa-router": "^7.4.0" "seneca": "^3.4.3" "seneca-web": "^2.2.0" "seneca-web-adapter-koa2": "^1.1.0" "amqplib": "^0.5.2" "winston": "^2.4.2" "mongoose": "^5.1.2"
FEATURES
- 模式匹配作服务间调用：略微不一样于SpringCloud服务发现（http协议、IP + PORT模式），它使用更加灵活的模式匹配（Patrun模块）原则去进行微服务间的调用
- 接入koa2对C端提供RESTFUl API
- 插件：更灵活编写小而微的可复用模块
- seneca内置日志输出
- 第三方日志库比较winston（选用）、bunyan、log4js
- RabbitMQ消息队列
- PM2：node服务部署（服务集群）、管理与监控
- PM2：自动化部署
- PM2集成docker
- 请求追踪（重建用户请求流程）
- 梳理Consul 服务注册与发现基本逻辑
- 框架集成node-consul
- mongodb持久化存储
- 结合seneca与consul的路由服务中间件（可支持多个相同名字服务集群路由，经过$$version区别）
- 支持流处理（文件上传/下载等）
- jenkins自动化部署
- nginx负载均衡
- 持续集成方案
- redis缓存
- prisma提供GraphQL接口
模式匹配（Patrun模块）
index.js(accout-server/src/index.js)nginx
const seneca = require('seneca')()
seneca.use('cmd:login', (msg, done) => {
const { username, pass } = msg
if (username === 'asd' && pass === '123') {
return done(null, { code: 1000 })
}
return done(null, { code: 2100 })
})
const Promise = require('bluebird')
const act = Promise.promisify(seneca.act, { context: 'seneca' })
act({
cmd: 'login',
username: 'asd',
pass: '123'
}).then(res => {
console.log(res)
}).catch(err => {
console.log(err)
})
执行后git
{ code: 1000 }
{"kind":"notice","notice":"hello seneca k5i8j1cvw96h/1525589223364/10992/3.4.3/-","level":"info","seneca":"k5i8j1cvw96h/1525589223364/10992/3.4.3/-","when":1525589223563}
seneca.add方法，添加一个action pattern到Seneca实例中，它有三个参数：github
-
pattern: 用于Seneca中JSON的消息匹配模式，对象或格式化字符串 -
sub_pattern: 子模式，优先级低于主模式（可选） -
action: 当匹配成功后的动做函数
seneca.act方法，执行Seneca实例中匹配成功的动做，它也有两个参数：web
-
msg: JSON消息 -
sub_pattern: 子消息，优先级低于主消息（可选） -
response: 用于接收服务调用结果
seneca.use方法，为Seneca实例添加一个插件，它有两个参数：（此处插件的原理和中间件有一些不一样）redis
-
func: 插件执行方法 -
options: 插件所需options（可选）
核心是利用JSON对象进行模式匹配。这个JSON对象既包含某个微服务所须要调取另外一个微服务的特征，同时也包含传参。和Java微服务发现有些相似不过是用模式代替ip+port，目前为止模式是彻底能够实现服务发现功能，可是否更加灵活还有待去挖掘。mongodb
所需注意的点docker
- 各微服务之间模式需经过设计来区分
启动第一个微服务
index.js(config-server/src/index.js)shell
const seneca = require('seneca')()
const config = {
SUCCESS_NORMAL_RES: {
code: 1000,
desc: '服务端正常响应'
}}
seneca.add('$target$:config-server', (msg, done) => {
return done(null, config)
}).listen(10011)
运行此脚本后可在浏览器中输入http://localhost:10011/act?cmd=config发起请求获取全局配置信息
OR
const seneca = require('seneca')()
const Promise = require('bluebird')
const act = Promise.promisify(seneca.act, { context: seneca })
seneca.client(10011)
act('$$target:config-server, default$:{msg:404}').then(res => {
console.log(res)
}).catch(err => {
console.log(err)
})
对内：多个微服务相互调用（关键）
noname-server
const seneca = require('seneca')()
seneca.add('$$target:account-server', (msg, done) => {
done(null, { seneca: '666' })
})
seneca.listen(10015)
config-server（同上）
call
const seneca = require('seneca')()
const Promise = require('blurebird')
const act = Promise.promisify(seneca.act, { context: seneca })
seneca.client({
port: '10011',
pin: '$$target:account-server'
})
seneca.client({
port: '10015',
pin: '$$target:noname-server'
})
act('$$target:account-server').then(res => {
console.log(res)
}).catch(err => {
console.log(err)
})
act('$$target:noname-server').then(res => {
console.log(res)
}).catch(err => {
console.log(err)
})
对外：提供REST服务（关键）
集成koa
const seneca = require('seneca')()
const Promise = require('bluebird')
const SenecaWeb = require('seneca-web')
const Koa = require('koa')
const Router = require('koa-router')
const app = new Koa()
const userModule = require('./modules/user.js')
// 初始化用户模块
seneca.use(userModule.init)
// 初始化seneca-web插件，并适配koa
seneca.use(SenecaWeb, {
context: Router(),
adapter: require('seneca-web-adapter-koa2'),
routes: [...userModule.routes]
})
// 将routes导出给koa app
seneca.ready(() => {
app.use(seneca.export('web/context')().routes())
})
app.listen(3333)
user模块
const $module = 'module:user'
let userCount = 3
const REST_Routes = [
{
prefix: '/user',
pin: `${$module},if:*`,
map: {
list: {
GET: true,
name: ''
},
load: {
GET: true,
name: '',
suffix: '/:id'
},
edit: {
PUT: true,
name: '',
suffix: '/:id'
},
create: {
POST: true,
name: ''
},
delete: {
DELETE: true,
name: '',
suffix: '/:id'
}
}
}
]
const db = {
users: [{
id: 1,
name: '甲'
}, {
id: 2,
name: '乙'
}, {
id: 3,
name: '丙'
}]
}
function user(options) {
this.add(`${$module},if:list`, (msg, done) => {
done(null, db.users)
})
this.add(`${$module},if:load`, (msg, done) => {
const { id } = msg.args.params
done(null, db.users.find(v => Number(id) === v.id))
})
this.add(`${$module},if:edit`, (msg, done) => {
let { id } = msg.args.params
id = +id
const { name } = msg.args.body
const index = db.users.findIndex(v => v.id === id)
if (index !== -1) {
db.users.splice(index, 1, {
id,
name
})
done(null, db.users)
} else {
done(null, { success: false })
}
})
this.add(`${$module},if:create`, (msg, done) => {
const { name } = msg.args.body
db.users.push({
id: ++userCount,
name
})
done(null, db.users)
})
this.add(`${$module},if:delete`, (msg, done) => {
let { id } = msg.args.params
id = +id
const index = db.users.findIndex(v => v.id === id)
if (index !== -1) {
db.users.splice(index, 1)
done(null, db.users)
} else {
done(null, { success: false })
}
})
}
module.exports = {
init: user,
routes: REST_Routes
}
vscode-restclient(vscode的restclient插件，用于发起RESTFUL请求)
### 1
POST http://localhost:3333/user HTTP/1.1
Content-Type: application/json
{
"name": "测试添加用户"
}
### delete
DELETE http://localhost:3333/user/2 HTTP/1.1
### PUT
PUT http://localhost:3333/user/2 HTTP/1.1
Content-Type: application/json
{
"name": "测试修改用户信息"
}
### GET
GET http://localhost:3333/user HTTP/1.1
### GET
GET http://localhost:3333/user/3 HTTP/1.1
seneca内置日志输出
可在构造函数中传入配置，log属性能够控制日志级别
例1：传字符串
require('seneca')({
// quiet silent any all print standard test
log: 'all'
})
例2：传对象
require('seneca')({
log: {
// none debug+ info+ warn+
level: 'debug+'
},
// 设置为true时，seneca日志功能会encapsulate senecaId,senecaTag,actId等字段后输出（通常为两字符）
short: true
})
建议例2代码，由于seneca-web-adapter-koa2插件打印的日志level为debug，利于作web接口访问日志记录。
winston日志模块
Logger.js
const { createLogger, format, transports } = require('winston')
const { combine, timestamp, label, printf } = format
const logger = createLogger({
level: 'info',
format: combine(
label({label: 'microservices'}),
timestamp(),
printf(info => {
return `${info.timestamp} [${info.label}] ${info.level}: ${info.message}`
})
),
transports: [ new transports.Console() ]
})
// highest to lowest
const levels = {
error: 0,
warn: 1,
info: 2,
verbose: 3,
debug: 4,
silly: 5
}
module.exports = logger
日志输出格式
2018-05-17T14:43:28.330Z [microservices] info: 接收到rpc客户端的调用请求 2018-05-17T14:43:28.331Z [microservices] warn: warn message 2018-05-17T14:43:28.331Z [microservices] error: error message
RabbitMQ消息队列服务
1. 单任务单consumer，生产者消费者模式
producer.js
// 建立一个amqp对等体
const amqp = require('amqplib/callback_api')
amqp.connect('amqp://localhost', (err, conn) => {
conn.createChannel((err, ch) => {
const q = 'taskQueue1'
const msg = process.argv.slice(2).join(' ') || 'hello world'
// 为方式RabbitMQ退出或者崩溃时重启后丢失队列信息，这里配置durable:true（同时在消费者脚本中也要配置durable:true）后，
ch.assertQueue(q, { durable: true })
// 这里配置persistent:true，经过阅读官方文档，我理解为当程序重启后，会断点续传以前未send完成的数据消息。（但此功能并不可靠，由于不会为全部消息执行同步IO，会缓存在cache并在某个恰当时机write到disk）
ch.sendToQueue(q, Buffer.from(msg), { persistent: true })
setTimeout(() => {
conn.close(); process.exit(0)
}, 100)
})
})
// 建立一个amqp对等体
const amqp = require('amqplib/callback_api')
amqp.connect('amqp://localhost', (err, conn) => {
conn.createChannel((err, ch) => {
const q = 'taskQueue1'
// 为方式RabbitMQ退出或者崩溃时重启后丢失队列信息，这里配置durable:true（同时在消费者脚本中也要定义durable:true）后，
ch.assertQueue(q, { durable: true })
ch.prefetch(1)
console.log(" [*] Waiting for messages in %s. To exit press CTRL+C", q)
ch.consume(q, msg => {
const secs = msg.content.toString().split('.').length - 1
console.log(" [x] Received %s", msg.content.toString())
setTimeout(() => {
console.log(" [x] Done")
ch.ack(msg)
}, secs * 1000)
})
// noAck配置（默认为false）代表consumer是否须要在处理完后反馈ack给producer，若是设置为true，则RabbitMQ服务若是将任务send至此consumer后不关心任务实际处理结果，send任务后直接标记已完成；不然，RabbiMQ获得ack反馈后才标记为已完成，若是一直未收到ack默认会一直等待ack而后标记，另外若是接收到nack或者该consumer进程退出则继续dispatcher任务
})
})
检验过程
- 执行rabbitmqctl list_queues查看当前队列
Timeout: 60.0 seconds ... Listing queues for vhost / ...
- node producer.js（rabbitMQ执行过程为会先建立一个匿名exchange，一个指定queue而后将queue与该匿名exchange绑定）
- rabbitmqctl list_bindings
Listing bindings for vhost /...
exchange taskQueue1 queue taskQueue1 []
- rabbitmqctl list_queues
Timeout: 60.0 seconds ... Listing queues for vhost / ... taskQueue1 1
- node consumer.js
Waiting for messages in taskQueue1. To exit press CTRL+C [x] Received hello world [x] Done
- rabbitmqctl list_queues
Timeout: 60.0 seconds ... Listing queues for vhost / ... taskQueue1 0
知识点
- 生产者消费者模式（一个生产者的消息在同一时间只交由一个消费者处理）
- ACK机制（rabbitmq的确认机制）
- 建立队列{durable:true}以及向队列发送消息{persistent:true}（消息持久化存储，但不彻底能保证，好比当某消息未从缓存中写到磁盘中而程序崩溃时则会丢失）
- Round-robin Dispatch（公平分发）
- 处理窗口控制（prefetch来控制分发窗口）
- 异步多任务处理机制（好比一个大任务分解，分而治之）
- 整个消息流流程（某个生产者进程 -> 匿名exchange -> 经过binding -> 指定queue -> 某一个消费者进程）
2. 单任务多consumer，发布/订阅模式（全消息模型）
publisher.js
const amqp = require('amqplib/callback_api')
amqp.connect('amqp://localhost', (err, conn) => {
conn.createChannel((err, ch) => {
const ex = 'logs'
const msg = process.argv.slice(2).join(' ') || 'Hello World!'
// ex为exchange名称（惟一）
// 模式为fanout
// 不对消息持久化存储
ch.assertExchange(ex, 'fanout', { durable: false })
// 第二个参数为指定某一个binding，如为空则由RabbitMQ随机指定
ch.publish(ex, '', Buffer.from(msg))
console.log(' [x] Send %s', msg)
})
setTimeout(() => {
conn.close()
process.exit(0)
}, 100)
})
subscriber.js
const amqp = require('amqplib/callback_api')
amqp.connect('amqp://localhost', (err, conn) => {
conn.createChannel((err, ch) => {
const ex = 'logs'
// ex -> exchange是发布/订阅消息的载体，
// fanout -> 分发消息的模式，fanout，direct，topic，headers
// durable设置为false下降一些可靠性，提升性能，由于不须要磁盘IO持久化存储消息，另外
ch.assertExchange(ex, 'fanout', { durable: false })
// 使用匿名（也就是RabbitMQ自动生成随机名的queue）队列
// exclusive设置为true，便可以当其寄生的connection被close的时候自动deleted
ch.assertQueue('', { exclusive: true }, (err, q) => {
console.log(" [*] Waiting for messages in %s. To exit press CTRL+C", q.queue)
// 绑定队列到某个exchange载体（监听某个exchange的消息）
// 第三个入参为binding key
ch.bindQueue(q.queue, ex, '')
// 消费即订阅某个exchange的消息并设置处理句柄
// 由于发布/订阅消息的模式就是非可靠性，只有当订阅者订阅才能收到相关的消息并且发布者不关心该消息的订阅者是谁以及处理结果如何，因此这里noAck会置为true
ch.consume(q.queue, (msg) => {
console.log(' [x] %s', msg.content.toString())
}, { noAck: true })
})
})
})
检验过程
rabbitmqctl stop_app;rabbitmqctl reset;rabbitmqctl start_app（清空以前测试使用的queues、echanges、bindings）
node subscriber.js
[*] Waiting for messages in amq.gen-lgNW51IeEfj9vt1yjMUuaw. To exit press CTRL+C
rabbitmqctl list_exchanges
Listing exchanges for vhost / ... logs fanout
rabbitmqctl list_bindings
Listing bindings for vhost /...
exchange amq.gen-jDbfwJR8TbSNJT2a2a83Og queue amq.gen-jDbfwJR8TbSNJT2a2a83Og []
logs exchange amq.gen-jDbfwJR8TbSNJT2a2a83Og queue []
node publisher.js tasks.........
[x] Send tasks......... // publiser.js [x] tasks......... // subscriber.js
知识点
- 发布/订阅模式（发布者将消息以一对多的形式发送给订阅者处理）
- noAck（此模式下推荐用非Ack机制，由于发布者每每不须要订阅者如何处理消息以及其结果）
- durable:false（此模式下推荐不须要作数据持久化存储，缘由如上）
- exchange的工做模式（即路由类型，fanout，direct，topic，headers等，下节会讲解到）
- 整个消息流流程（某个发布者进程 -> 指定exchange -> 经过binding以及工做模式 -> 某个或多个匿名queue即订阅者进程）
3. Direct Routing
exchange.js
module.exports = {
name: 'ex1',
type: 'direct',
option: {
durable: false
},
ranks: ['info', 'error', 'warning', 'severity']
}
direct-routing.js
const amqp = require('amqplib/callback_api')
const ex = require('./exchange')
amqp.connect('amqp://localhost', (err, conn) => {
conn.createChannel((err, ch) => {
ch.assertExchange(ex.name, ex.type, ex.options)
setTimeout(() => {
conn.close()
process.exit(0)
}, 0)
})
})
subscriber.js
const amqp = require('amqplib/callback_api')
const ex = require('./exchange')
amqp.connect('amqp://localhost', (err, conn) => {
conn.createChannel((err, ch) => {
const ranks = ex.ranks
ranks.forEach(rank => {
// 声明一个非匿名queue
ch.assertQueue(`${rank}-queue`, { exclusive: false }, (err, q) => {
ch.bindQueue(q.queue, ex.name, rank)
ch.consume(q.queue, msg => {
console.log(" [x] %s: '%s'", msg.fields.routingKey, msg.content.toString());
}, { noAck: true })
})
})
})
})
publisher.js
const amqp = require('amqplib/callback_api')
const ex = require('./exchange')
amqp.connect('amqp://localhost', (err, conn) => {
conn.createChannel((err, ch) => {
const ranks = ex.ranks
ranks.forEach(rank => {
ch.publish(ex.name, rank, Buffer.from(`${rank} logs...`))
})
setTimeout(() => {
conn.close()
process.exit(0)
}, 0)
})
})
检验过程
rabbitmqctl stop_app;rabbitmqctl reset;rabbitmqctl start_app（清空以前测试使用的queues、echanges、bindings）
node direct-routing.jsrabbitmqctl list_exchanges
Listing exchanges for vhost / ...
amq.headers headers
ex1 direct
amq.fanout fanout
amq.rabbitmq.trace topic
amq.topic topic
direct
amq.direct direct
amq.match headers
node subscriber.jsrabbitmqctl list_queues
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
severity-queue 0
error-queue 0
info-queue 0
warning-queue 0
Listing bindings for vhost /...
exchange error-queue queue error-queue []
exchange info-queue queue info-queue []
exchange severity-queue queue severity-queue []
exchange warning-queue queue warning-queue []
ex1 exchange error-queue queue error []
ex1 exchange info-queue queue info []
ex1 exchange severity-queue queue severity []
ex1 exchange warning-queue queue warning []
node publisher.js
[x] info: 'info logs...' [x] error: 'error logs...' [x] severity: 'severity logs...' [x] warning: 'warning logs...'
知识点
- 路由key，用于exchange的direct工做模式下消息的路由
- 每当assertQueue时，该queue会在以queue名称看成路由key绑定到匿名exchange
- 可用于日志不一样级别的log处理
4. Topic Routing
exchange.js
module.exports = {
name: 'ex2',
type: 'topic',
option: {
durable: false
},
ranks: ['info', 'error', 'warning', 'severity']
}
topic-routing.js
const amqp = require('amqplib/callback_api')
const exchangeConfig = require('./exchange')
amqp.connect('amqp://localhost', (err, conn) => {
conn.createChannel((err, ch) => {
ch.assertExchange(exchangeConfig.name, exchangeConfig.type, exchangeConfig.option)
setTimeout(() => {
conn.close()
process.exit(0)
}, 0)
})
})
subscriber.js
const amqp = require('amqplib/callback_api')
const exchangeConfig = require('./exchange')
amqp.connect('amqp://localhost', (err, conn) => {
conn.createChannel((err, ch) => {
const args = process.argv.slice(2)
const keys = (args.length > 0) ? args : ['anonymous.info']
console.log(' [*] Waiting for logs. To exit press CTRL+C');
keys.forEach(key => {
ch.assertQueue('', { exclusive: true }, (err, q) => {
console.log(` [x] Listen by routingKey ${key}`)
ch.bindQueue(q.queue, exchangeConfig.name, key)
ch.consume(q.queue, msg => {
console.log(" [x] %s:'%s'", msg.fields.routingKey, msg.content.toString());
}, { noAck: true })
})
})
})
})
publisher.js
const amqp = require('amqplib/callback_api')
const exchangeConfig = require('./exchange')
amqp.connect('amqp://localhost', (err, conn) => {
conn.createChannel((err, ch) => {
const args = process.argv.slice(2)
const key = (args.length > 1) ? args[0] : 'anonymous.info'
const msg = args.slice(1).join(' ') || 'hello world'
ch.publish(exchangeConfig.name, key, Buffer.from(msg))
setTimeout(() => {
conn.close()
process.exit(0)
}, 0)
})
})
检验过程
rabbitmqctl stop_app;rabbitmqctl reset;rabbitmqctl start_app（清空以前测试使用的queues、echanges、bindings）
node topic-routing.js
Listing exchanges for vhost / ...
amq.fanout fanout
amq.rabbitmq.trace topic
amq.headers headers
amq.match headers
ex2 topic
direct
amq.topic topic
amq.direct direct
node subscriber.js "#.info" "*.error"
[*] Waiting for logs. To exit press CTRL+C [x] Listen by routingKey #.info [x] Listen by routingKey *.error
- node publisher.js "account-server.info" "用户服务测试"
- node publisher.js "config-server.info" "配置服务测试"
- node publisher.js "config-server.error" "配置服务出错"
[x] account-server.info:'用户服务测试' [x] config-server.info:'配置服务测试' [x] config-server.error:'配置服务出错'
知识点
- key最长为255字节
-
#可匹配0或多个单词，*可精确匹配1个单词
5. RPC
rpc_server.js
const amqp = require('amqplib/callback_api')
const logger = require('./Logger')
let connection = null
amqp.connect('amqp://localhost', (err, conn) => {
connection = conn
conn.createChannel((err, ch) => {
const q = 'account_rpc_queue'
ch.assertQueue(q, { durable: true })
ch.prefetch(2)
ch.consume(q, msg => {
let data = {}
let primitiveContent = msg.content.toString()
try {
data = JSON.parse(primitiveContent)
} catch (e) {
logger.error(new Error(e))
}
logger.info('接收到rpc客户端的调用请求')
if (msg.properties.correlationId === '10abc') {
logger.info(primitiveContent)
const uid = Number(data.uid) || -1
let r = getUserById(uid)
ch.sendToQueue(msg.properties.replyTo, Buffer.from(JSON.stringify(r)), { persistent: true })
ch.ack(msg)
} else {
logger.info('不匹配的调用请求')
}
})
})
})
function getUserById (uid) {
let result = ''
if (uid === +uid && uid > 0) {
result = {
state: 1000,
msg: '成功',
data: {
uid: uid,
name: '小强',
sex: 1
}
}
} else {
result = {
state: 2000,
msg: '传参格式错误'
}
}
return result
}
process.on('SIGINT', () => {
logger.warn('SIGINT')
connection && connection.close()
process.exit(0)
})
rpc_client.js
const amqp = require('amqplib/callback_api')
amqp.connect('amqp://localhost', (err, conn) => {
conn.createChannel((err, ch) => {
const q = 'account_rpc_queue'
const callback = 'callback_queue'
ch.assertQueue(callback, { durable: true })
ch.consume(callback, msg => {
const result = msg.content.toString()
console.log(`接收到回调的消息啦！`)
console.log(result)
ch.ack(msg)
setTimeout(() => {
conn.close()
process.exit(0)
}, 0)
})
ch.assertQueue(q, { durable: true })
const msg = {
uid: 2
}
ch.sendToQueue(q, Buffer.from(JSON.stringify(msg)), {
persistent: true,
correlationId: '10abc',
replyTo: 'callback_queue'
})
})
})
检验过程
node rpc_server.js
rabbitmqctl list_queues
Timeout: 60.0 seconds ... Listing queues for vhost / ... account_rpc_queue 0
node rpc_client.js
rpc_client的CLI打印
接收到回调的消息啦！
{"state":1000,"msg":"成功","data":{"uid":2,"name":"小强","sex":1}}
rpc_server的CLI打印
接收到rpc客户端的调用请求
{ uid: 2 }
PM2：node服务部署（服务集群）、管理与监控
启动
pm2 start app.js
-
-w --watch：监听目录变化，如变化则自动重启应用 -
--ignore-file：监听目录变化时忽略的文件。如pm2 start rpc_server.js --watch --ignore-watch="rpc_client.js" -
-n --name：设置应用名字，可用于区分应用 -
-i --instances：设置应用实例个数，0与max相同 -
-f --force： 强制启动某应用，经常用于有相同应用在运行的状况 -
-o --output <path>：标准输出日志文件的路径 -
-e --error <path>：错误输出日志文件的路径 -
--env <path>：配置环境变量
如pm2 start rpc_server.js -w -i max -n s1 --ignore-watch="rpc_client.js" -e ./server_error.log -o ./server_info.log
在cluster-mode，也就是-i max下，日志文件会自动在后面追加-${index}保证不重复
其余简单且经常使用命令
pm2 stop app_name|app_idpm2 restart app_name|app_idpm2 delete app_name|app_idpm2 show app_name|app_id OR pm2 describe app_name|app_idpm2 listpm2 monitpm2 logs app_name|app_id --lines <n> --err
Graceful Stop
pm2 stop app_name|app_id
process.on('SIGINT', () => {
logger.warn('SIGINT')
connection && connection.close()
process.exit(0)
})
当进程结束前，程序会拦截SIGINT信号从而在进程即将被杀掉前去断开数据库链接等等占用内存的操做后再执行process.exit()从而优雅的退出进程。（如在1.6s后进程还未结束则继续发送SIGKILL信号强制进程结束）
Process File
ecosystem.config.js
const appCfg = {
args: '',
max_memory_restart: '150M',
env: {
NODE_ENV: 'development'
},
env_production: {
NODE_ENV: 'production'
},
// source map
source_map_support: true,
// 不合并日志输出，用于集群服务
merge_logs: false,
// 经常使用于启动应用时异常，超时时间限制
listen_timeout: 5000,
// 进程SIGINT命令时间限制，即进程必须在监听到SIGINT信号后必须在如下设置时间结束进程
kill_timeout: 2000,
// 当启动异常后不尝试重启，运维人员尝试找缘由后重试
autorestart: false,
// 不容许以相同脚本启动进程
force: false,
// 在Keymetrics dashboard中执行pull/upgrade操做后执行的命令队列
post_update: ['npm install'],
// 监听文件变化
watch: false,
// 忽略监听文件变化
ignore_watch: ['node_modules']
}
function GeneratePM2AppConfig({ name = '', script = '', error_file = '', out_file = '', exec_mode = 'fork', instances = 1, args = "" }) {
if (name) {
return Object.assign({
name,
script: script || `${name}.js`,
error_file: error_file || `${name}-err.log`,
out_file: out_file|| `${name}-out.log`,
instances,
exec_mode: instances > 1 ? 'cluster' : 'fork',
args
}, appCfg)
} else {
return null
}
}
module.exports = {
apps: [
GeneratePM2AppConfig({
name: 'client',
script: './rpc_client.js'
}),
GeneratePM2AppConfig({
name: 'server',
script: './rpc_server.js',
instances: 1
})
]
}
pm2 start ecosystem.config.js
避坑指南：processFile文件命名建议为*.config.js格式。不然后果自负。
监控
PM2：自动化部署
ssh准备
- ssh-keygen -t rsa -C 'qingf deployment' -b 4096
- 若是有多密钥、多用户状况，建议配置~/.ssh/config文件，格式相似以下
// 用不一样用户对不一样远程主机发起ssh请求时指定私钥 Host qingf.me User deploy IdentityFile ~/.ssh/qf_deployment_rsa // 设置为no可去掉首次登录(y/n)的选择 StrictHostKeyChecking no // 别名用法 Host deployment User deploy Hostname qingf.me IdentityFile ~/.ssh/qingf_deployment_rsa StrictHostKeyChecking no
- 将公钥复制到远程（通常为部署服务器）对应用户目录，好比/home/deploy/.ssh/authorized_keys文件（authorized_keys文件权限设置为600）
配置ecosystem.config.js
与上述apps同级增长deploy属性，以下
deploy: {
production: {
'user': 'deploy',
'host': 'qingf.me',
'ref': 'remotes/origin/master',
'repo': 'https://github.com/Cecil0o0/account-server.git',
'path': '/home/deploy/apps/account-server',
// 生命周期钩子，在ssh到远端以后setup操做以前执行
'pre-setup': '',
// 生命周期钩子，在初始化设置即git pull以后执行
'post-setup': 'ls -la',
// 生命周期钩子，在远端git fetch origin以前执行
'pre-setup': '',
// 生命周期钩子，在远端git修改HEAD指针到指定ref以后执行
'post-deploy': 'npm install && pm2 startOrRestart deploy/ecosystem.config.js --env production',
// 如下这个环境变量将注入到全部app中
"env" : {
"NODE_ENV": "test"
}
}
}
tip：please make git working directory clean first!
而后前后执行如下两条命令（注意config文件路径）
- pm2 deploy deploy/ecosystem.config.js production setup
- pm2 deploy deploy/ecosystem.config.js production
其余命令
pm2 deploy <configuration_file> <environment> <command>
Commands:
setup run remote setup commands
update update deploy to the latest release
revert [n] revert to [n]th last deployment or 1
curr[ent] output current release commit
prev[ious] output previous release commit
exec|run <cmd> execute the given <cmd>
list list previous deploy commits
[ref] deploy to [ref], the "ref" setting, or latest tag
推荐shell toolkit
请求追踪
如何？
- 在seneca.add以及seneca.act中使用seneca.fixedargs['tx$']值做为traceID标识处于某一条请求流程。另外seneca内置log系统会打印此值。
疑问？
seneca内置log系统如何作自定义日志打印？
舒适提示：请以正常的http请求开始，由于通过测试若是微服务自主发起act，其seneca.fixedargs['tx$']值不一样。
Consul 服务注册与发现
Consul是一个分布式集群服务注册发现工具，并具备健康检查、分级式KV存储、多数据中心等高级特性。
安装
- 可选择使用预编译的安装包
- 也可选择克隆源码后编译安装
基础使用
- 以开发模式快速启动服务模式代理并开启web界面访问http://localhost:8500
consul agent -dev -ui
- 编写服务定义文件
{
"service": {
// 服务名，稍后用于query服务
"name": "account-server",
// 服务标签
"tags": ["account-server"],
// 服务元信息
"meta": {
"meta": "for my service"
},
// 服务端口
"port": 3333,
// 不容许标签覆盖
"enable_tag_override": false,
// 脚本检测作health checks 与-enable-script-checks=true配合使用，有脚本模式、TCP模式、HTTP模式、TTL模式
"checks": [
{
"http": "http://localhost:3333/user",
"interval": "10s"
}
]
}
}
- query定义的account-server服务
curl http://localhost:8500/v1/catalog/service/account-server
[
{
"ID": "e66eb1ff-460c-e63f-b4ac-0cb42daed19c",
"Node": "haojiechen.local",
"Address": "127.0.0.1",
"Datacenter": "dc1",
"TaggedAddresses": {
"lan": "127.0.0.1",
"wan": "127.0.0.1"
},
"NodeMeta": {
"consul-network-segment": ""
},
"ServiceID": "account-server",
"ServiceName": "account-server",
"ServiceTags": [
"account-server"
],
"ServiceAddress": "",
"ServiceMeta": {
"meta": "for my service"
},
"ServicePort": 3333,
"ServiceEnableTagOverride": false,
"CreateIndex": 6,
"ModifyIndex": 6
}
]
生产级别使用（分布式集群）
某一个结点启动一个server模式代理，以下
consul agent -server -bootstrap-expect=1 \
-data-dir=/tmp/consul -node=agent-one -bind=valid extranet IP \
-enable-script-checks=true -config-dir=/usr/local/etc/consul.d
查看集群成员
consul members
Node Address Status Type Build Protocol DC Segment agent-one valid extranet IP:8301 alive server 1.1.0 2 dc1 <all>
另外一个结点启动一个client模式代理，以下
consul agent \
-data-dir=/tmp/consul -node=agent-two -bind=139.129.5.228 \
-enable-script-checks=true -config-dir=/usr/local/etc/consul.d
查看集群成员
consul members
Node Address Status Type Build Protocol DC Segment agent-two 139.129.5.228:8301 alive server 1.1.0 2 dc1 <all>
加入Cluster
consul join 139.129.5.228consul members
Node Address Status Type Build Protocol DC Segment agent-one valid extranet IP:8301 alive server 1.1.0 2 dc1 <all> agent-two 139.129.5.228:8301 alive server 1.1.0 2 dc1 <all>
集成node-consul
config.js
// 服务注册与发现
// https://github.com/silas/node-consul#catalog-node-services
'serverR&D': {
consulServer: {
type: 'consul',
host: '127.0.0.1',
port: 8500,
secure: false,
ca: [],
defaults: {
token: ''
},
promisify: true
},
bizService: {
name: 'defaultName',
id: 'defaultId',
address: '127.0.0.1',
port: 1000,
tags: [],
meta: {
version: '',
description: '注册集群'
},
check: {
http: '',
// check间隔时间(ex: 15s)
interval: '10s',
// check超时时间(ex: 10s)
timeout: '2s',
// 处于临界状态后自动注销服务的超时时间
deregistercriticalserviceafter: '30s',
// 初始化状态值为成功
status: 'passing',
// 备注
notes: '{"version":"111","microservice-port":1115}'
}
}
}
server-register.js
/*
* @Author: Cecil
* @Last Modified by: Cecil
* @Last Modified time: 2018-06-02 11:26:49
* @Description 微服务注册方法
*/
const defaultConf = require('../config')['serverR&D']
const { ObjectDeepSet, isString } = require('../helper/utils')
const Consul = require('consul')
const { generateServiceName, generateCheckHttp } = require('../helper/consul')
// 注册服务
function register({ consulServer = {}, bizService = {} } = {}) {
if (!bizService.name && isString(bizService.name)) throw new Error('name is invalid!')
if (bizService.port !== +bizService.port) throw new Error('port is invalid!')
if (!bizService.host && isString(bizService.host)) throw new Error('host is invalid!')
if (!bizService.meta.$$version) throw new Error('meta.$$version is invalid!')
if (!bizService.meta.$$microservicePort) throw new Error('meta.$$microservicePort is invalid!')
const consul = Consul(ObjectDeepSet(defaultConf.consulServer, consulServer))
const service = defaultConf.bizService
service.name = generateServiceName(bizService.name)
service.id = service.name
service.address = bizService.host
service.port = bizService.port
service.check.http = generateCheckHttp(bizService.host, bizService.port)
service.check.notes = JSON.stringify(bizService.meta)
return new Promise((resolve, reject) => {
consul.agent.service.list().then(services => {
// 检查主机+端口是否已被占用
Object.keys(services).some(key => {
if (services[key].Address === service.address && services[key].Port === service.port) {
throw new Error(`该服务集群endpoint[${service.address}, ${service.port}]已被占用！`)
}
})
// 注册集群服务
consul.agent.service.register(service).then(() => {
logger.info(`${bizService.name}服务已注册`)
resolve(services)
}).catch(err => {
console.log(err)
})
}).catch(err => {
throw new Error(err)
})
})
}
module.exports = class ServerRegister {
constructor() {
this.register = register
}
}
验证
保证runtime中存在consul和mongodb服务后，clone该仓库Demo，cd到工程根目录下，运行node src便可。
框架集成node-consul
server-register.js
/*
* @Author: Cecil
* @Last Modified by: Cecil
* @Last Modified time: 2018-06-02 13:58:22
* @Description 微服务注册方法
*/
const defaultConf = require('../config')['serverR&D']
const { ObjectDeepSet, isString } = require('../helper/utils')
const Consul = require('consul')
const { generateServiceName, generateCheckHttp } = require('../helper/consul')
const logger = new (require('./logger'))().generateLogger()
// 注册服务方法定义
function register({ consulServer = {}, bizService = {} } = {}) {
if (!bizService.name && isString(bizService.name)) throw new Error('name is invalid!')
if (bizService.port !== +bizService.port) throw new Error('port is invalid!')
if (!bizService.host && isString(bizService.host)) throw new Error('host is invalid!')
if (!bizService.meta.$$version) throw new Error('meta.$$version is invalid!')
if (!bizService.meta.$$microservicePort) throw new Error('meta.$$microservicePort is invalid!')
const consul = Consul(ObjectDeepSet(defaultConf.consulServer, consulServer))
const service = defaultConf.bizService
service.name = generateServiceName(bizService.name)
service.id = service.name
service.address = bizService.host
service.port = bizService.port
service.check.http = generateCheckHttp(bizService.host, bizService.port)
service.check.notes = JSON.stringify(bizService.meta)
return new Promise((resolve, reject) => {
consul.agent.service.list().then(services => {
// 检查主机+端口是否已被占用
Object.keys(services).some(key => {
if (services[key].Address === service.address && services[key].Port === service.port) {
throw new Error(`该服务集群endpoint[${service.address}, ${service.port}]已被占用！`)
}
})
// 注册集群服务
consul.agent.service.register(service).then(() => {
logger.info(`${bizService.name}服务注册成功`)
resolve(services)
}).catch(err => {
console.log(err)
})
}).catch(err => {
throw new Error(err)
})
})
}
module.exports = class ServerRegister {
constructor() {
this.register = register
}
}
account-server/src/index.js
const vastify = require('vastify')
const version = require('../package.json').version
const microservicePort = 10015
const httpPort = 3333
// 注册服务
vastify.ServerRegister.register({
bizService: {
name: 'account-server',
host: '127.0.0.1',
port: httpPort,
meta: {
$$version: version,
$$microservicePort: microservicePort
}
}
})
Mongodb持久化存储
- 框架使用mongoose作mongoClient，固然你也能够选用原生nodejs mongoClient。
改造以前的user模块，偷个懒就不贴代码了，具体请查看Demo
结合seneca以及consul的路由服务中间件
microRouting.js
/*
* @Author: Cecil
* @Last Modified by: Cecil
* @Last Modified time: 2018-06-02 16:22:02
* @Description 微服务内部路由中间件，暂不支持自定义路由匹配策略
*/
'use strict'
const Consul = require('consul')
const defaultConf = require('../config')
const { ObjectDeepSet, isNumber } = require('../helper/utils')
const { getServiceNameByServiceKey, getServiceIdByServiceKey } = require('../helper/consul')
const logger = new (require('../tools/logger'))().generateLogger()
const { IPV4_REGEX } = require('../helper/regex')
let services = {}
let consul = null
/**
* @author Cecil0o0
* @description 同步consul服务中心的全部可用服务以及对应check并组装成对象以方便取值
*/
function syncCheckList () {
return new Promise((resolve, reject) => {
consul.agent.service.list().then(allServices => {
if (Object.keys(allServices).length > 0) {
services = allServices
consul.agent.check.list().then(checks => {
Object.keys(checks).forEach(key => {
allServices[getServiceIdByServiceKey(key)]['check'] = checks[key]
})
resolve(services)
}).catch(err => {
throw new Error(err)
})
} else {
const errmsg = '未发现可用服务'
logger.warn(errmsg)
reject(errmsg)
}
}).catch(err => {
throw new Error(err)
})
})
}
function syncRoutingRule(senecaInstance = {}, services = {}) {
Object.keys(services).forEach(key => {
let service = services[key]
let name = getServiceNameByServiceKey(key)
let $$addr = service.Address
let $$microservicePort = ''
let $$version = ''
try {
let base = JSON.parse(service.check.Notes)
$$microservicePort = base.$$microservicePort
$$version = base.$$version
} catch (e) {
logger.warn(`服务名为${serviceName}。该服务check.Notes为非标准JSON格式，程序已忽略。请检查服务注册方式（请确保调用ServerRegister的register来注册服务）`)
}
if (IPV4_REGEX.test($$addr) && isNumber($$microservicePort)) {
if (service.check.Status === 'passing') {
senecaInstance.client({
host: $$addr,
port: $$microservicePort,
pin: {
$$version,
$$target: name
}
})
} else {
logger.warn(`${$$target}@${$$version || '无'}服务处于critical，所以没法使用`)
}
} else {
logger.warn(`主机（${$$addr}）或微服务端口号（${$$microservicePort}）有误，请检查`)
}
})
}
function startTimeInterval() {
setInterval(syncCheckList, defaultConf.routing.servicesRefresh)
}
function microRouting(consulServer) {
var self = this
consul = Consul(ObjectDeepSet(defaultConf['serverR&D'].consulServer, consulServer))
syncCheckList().then(services => {
syncRoutingRule(self, services)
})
}
module.exports = microRouting
在保证有consul与mongodb的runtime后，请结合这两个config-server，account-server Demo进行测试。
[未完待续....]
相关文章
- 1. nodejs微服务框架解决方案
- 2. 微服务详解（二）：解决方案
- 3. nodejs微服务健康检查方案
- 4. 前端微服务+React解决方案。
- 5. 微服务监控解决方案
- 6. 微服务解决方案调研
- 7. 微服务解决方案_微服务为您提供正确的解决方案
- 8. 微服务之：微服务架构带来的问题及解决方案
- 9. 微服务架构(三)：微服务重构应用及IBM解决方案
- 10. 微服务下分布式事务解决方案
- 更多相关文章...
- • SVN 解决冲突 - SVN 教程
- • 启动MySQL服务 - MySQL教程
- • Spring Cloud 微服务实战(三) - 服务注册与发现
- • 常用的分布式事务解决方案
相关标签/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
最新文章
- 1. 跳槽面试的几个实用小技巧，不妨看看！
- 2. Mac实用技巧 |如何使用Mac系统中自带的预览工具将图片变成黑白色?
- 3. Mac实用技巧 |如何使用Mac系统中自带的预览工具将图片变成黑白色?
- 4. 如何使用Mac系统中自带的预览工具将图片变成黑白色?
- 5. Mac OS非兼容Windows软件运行解决方案——“以VMware & Microsoft Access为例“
- 6. 封装 pyinstaller -F -i b.ico excel.py
- 7. 数据库作业三ER图待完善
- 8. nvm安装使用低版本node.js(非命令安装)
- 9. 如何快速转换图片格式
- 10. 将表格内容分条转换为若干文档
欢迎关注本站公众号,获取更多信息
相关文章
- 1. nodejs微服务框架解决方案
- 2. 微服务详解（二）：解决方案
- 3. nodejs微服务健康检查方案
- 4. 前端微服务+React解决方案。
- 5. 微服务监控解决方案
- 6. 微服务解决方案调研
- 7. 微服务解决方案_微服务为您提供正确的解决方案
- 8. 微服务之：微服务架构带来的问题及解决方案
- 9. 微服务架构(三)：微服务重构应用及IBM解决方案
- 10. 微服务下分布式事务解决方案