俊瑶先森

发表于 更新于 阅读次数：

概述

最近在用nestjs重新做RBAC构架。
本次完整实践nestjs的注册登陆，到jwt授权认证，到guard守卫拦截验证，到strategy的JWT策略。
和Spring Boot的Shiro对比下来，还是nestjs的写着舒服。

实践

auth.controller.ts
注册登陆的controller实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
import { Body, Controller, Post } from '@nestjs/common';
import { ApiOkResponse, ApiOperation, ApiTags } from '@nestjs/swagger';
import { Result } from 'src/common/utils/result';
import { CreateUserDto } from '../user/dto/create.dto';
import { LoginUserDto } from '../user/dto/login.dto';
import { UserEntity } from '../user/user.entity';
import { UserService } from '../user/user.service';
import { AuthService } from './auth.service';

@ApiTags('登录注册')
@Controller('v1/auth')
export class AuthController {
  constructor(
    private readonly userService: UserService,
    private readonly authService: AuthService
  ) { }

  @Post('register')
  @ApiOperation({ summary: '用户注册' })
  @ApiOkResponse({ type: UserEntity })
  async create(@Body() user: CreateUserDto): Promise<Result> {
    console.log('user', user)
    const res = await this.userService.create(user)
    return Result.ok(res)
  }

  @Post('login')
  @ApiOperation({ summary: '登录' })
  async login(@Body() dto: LoginUserDto): Promise<Result> {
    const res = await this.userService.login(dto.account, dto.password)
    return Result.ok(res)
  }
}

user.service.ts
注册登陆及JWT的service实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
import { HttpException, HttpStatus, Injectable, NotFoundException } from '@nestjs/common';
import { InjectRepository } from '@nestjs/typeorm';
import { Like, Repository, UpdateResult } from 'typeorm';
import { UserEntity } from './user.entity';
import { Result } from 'src/common/utils/result';
import { CreateUserDto } from './dto/create.dto';
import { QueryUserDto } from './dto/query.dto';
import { classToPlain, plainToClass } from 'class-transformer';
import { RedisService } from 'nestjs-redis';
import { ConfigService } from '@nestjs/config';
import { genSalt, hash, compare } from 'bcrypt'
import { JwtService } from '@nestjs/jwt';
import { UpdateUserDto } from './dto/update.dto';
@Injectable()
export class UserService {
  constructor(
    @InjectRepository(UserEntity)
    private readonly userRep: Repository<UserEntity>,
    private readonly redisService: RedisService,
    private readonly config: ConfigService,
    private readonly jwtService: JwtService,
  ) { }

  async create(dto: CreateUserDto): Promise<UserEntity | Result> {
    console.log(dto)
    const existing = await this.findByUsername(dto.username)
    if (existing) throw new HttpException('账号已存在，请调整后重新注册！', HttpStatus.NOT_ACCEPTABLE);
    const salt = await genSalt()
    dto.password = await hash(dto.password, salt)
    const user = plainToClass(UserEntity, { salt, ...dto }, { ignoreDecorators: true })
    console.log('user', user)
    const res = await this.userRep.save(user)
    return res
  }

  // 登录
  async login(account: string, password: string): Promise<object | Result> {
    const user = await this.findByUsername(account)
    console.log("user", user)
    if (!user) throw new HttpException('账号或密码错误', HttpStatus.NOT_FOUND);
    console.log('账号', account)
    console.log('密码', password)
    console.log('加密的密码', user.password)
    const checkPassword = await compare(password, user.password)
    console.log('是否一致', checkPassword)
    if (!checkPassword) throw new HttpException('账号或密码错误', HttpStatus.NOT_FOUND);
    // 生成 token
    const data = this.genToken({ id: user.id })
    return data
  }

  // 根据ID查找
  async findById(id: number): Promise<UserEntity> {
    const res = await this.userRep.findOne(id)
    if (!res) {
      throw new NotFoundException()
    }
    return res
  }

  // 根据用户名查找
  async findByUsername(username: string): Promise<UserEntity> {
    return await this.userRep.findOne({ username })
  }

  // 生成 token
  genToken(payload: { id: number }): Record<string, unknown> {
    const accessToken = `Bearer ${this.jwtService.sign(payload)}`
    const refreshToken = this.jwtService.sign(payload, { expiresIn: this.config.get('jwt.refreshExpiresIn') })
    return { accessToken, refreshToken }
  }

  // 刷新 token
  refreshToken(id: number): string {
    return this.jwtService.sign({ id })
  }

  // 校验 token
  verifyToken(token: string): number {
    try {
      if (!token) return 0
      const id = this.jwtService.verify(token.replace('Bearer ', ''))
      return id
    } catch (error) {
      return 0
    }
  }

  // 根据JWT解析的ID校验用户
  async validateUserByJwt(payload: { id: number }): Promise<UserEntity> {
    return await this.findById(payload.id)
  }
}

jwt-auth.guard.ts
守卫实现，拦截accessToken和refreshToken,过期进行续签。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
import { Injectable, CanActivate, ExecutionContext, UnauthorizedException } from '@nestjs/common';
import { AuthGuard } from '@nestjs/passport';
import { Observable } from 'rxjs';
import { UserService } from 'src/modules/user/user.service';
@Injectable()
export class JwtAuthGuard extends AuthGuard('jwt') {
  constructor(
    private readonly userService: UserService,
    ) {
    super()
  }
  async canActivate(
    context: ExecutionContext,
  ): Promise<boolean> {
    const req = context.switchToHttp().getRequest()
    const res = context.switchToHttp().getResponse()
    try {
      const accessToken = req.get('Authorization')
      if (!accessToken) throw new UnauthorizedException('请先登录')

      const atUserId = this.userService.verifyToken(accessToken)
      if (atUserId) return this.activate(context)
      console.log(req.user)
      const refreshToken = req.get('RefreshToken')
      const rtUserId = this.userService.verifyToken(refreshToken)
      if (!rtUserId) throw new UnauthorizedException('当前登录已过期，请重新登录')
      const user = await this.userService.findById(rtUserId)
      if (user) {
        const tokens = this.userService.genToken({ id: rtUserId })
        // request headers 对象 prop 属性全自动转成小写，
        // 所以 获取 request.headers['authorization'] 或 request.get('Authorization')
        // 重置属性 request.headers[authorization] = value
        req.headers['authorization'] = tokens.accessToken
        req.headers['refreshtoken'] = tokens.refreshToken
        // 在响应头中加入新的token，客户端判断响应头有无 Authorization 字段，有则重置
        res.header('Authorization', tokens.accessToken)
        res.header('RefreshToken', tokens.refreshToken)
        // 将当前请求交给下一级
        return this.activate(context)
      } else {
        throw new UnauthorizedException('用户不存在')
      }
    } catch (error) {
      // Logger
      return false
    }
  }

  async activate(context: ExecutionContext): Promise<boolean> {
    return super.canActivate(context) as Promise<boolean>
  }
}

jwt.strategy.ts
JWT解析后进行校验

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

import { Injectable, UnauthorizedException } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { PassportStrategy } from '@nestjs/passport';
import { ExtractJwt, Strategy } from 'passport-jwt';
import { UserService } from 'src/modules/user/user.service';

@Injectable()
export class JwtStrategy extends PassportStrategy(Strategy) {
  constructor(
    private readonly userService: UserService,
    private readonly config: ConfigService
    ) {
    super({
      jwtFromRequest: ExtractJwt.fromAuthHeaderAsBearerToken(),
      secretOrKey: config.get('jwt.secretkey'),
    })
  }

  async validate(payload: any) {
    const user = await this.userService.validateUserByJwt(payload)
    // 如果有用户信息，代表 token 没有过期，没有则 token 已失效
    if (!user) throw new UnauthorizedException()
    return user
  }
}

发表于 更新于 阅读次数：

概述

根据 ElasticStack-安装篇 安装好logstash，我们开始进行配置和同步数据。
本次实践通过logstash同步mysql到es。

配置

进入logstash容器进行安装logstash-input-jdbc

1
docker exec -it docker_logstash bash

安装logstash-input-jdbc

1
./bin/logstash-plugin install logstash-input-jdbc

下载 mysql-connector-java-8.0.24.jar
放入/data/dockers/logstash/config/mysql

在文件目录/data/dockers/logstash/config/conf.d下创建jdbc.conf文件，进行mysql数据到es的配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
input{
     jdbc{
          # 连接数据库
          jdbc_connection_string => "jdbc:mysql://47.119.168.111:3306/fob?serverTimezone=Asia/Shanghai&characterEncoding=utf8&useSSL=false"
          jdbc_user => "root"
          jdbc_password => "XXXXX"
          # 连接数据库的驱动包
          jdbc_driver_library => "/usr/share/logstash/config/mysql/mysql-connector-java-8.0.24.jar"
          jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
          jdbc_paging_enabled => "true"
          jdbc_page_size => "50000"
          codec => plain { charset => "UTF-8" }

          # 数据追踪
          # 追踪的字段
          tracking_column => "updated_at"
          # 上次追踪的元数据存放位置
          last_run_metadata_path => "/usr/share/logstash/config/lastrun/logstash_jdbc_last_run"
          # 设置时区
          jdbc_default_timezone => "Asia/Shanghai"  
          # sql 文件地址
          # statement_filepath => ""
          # sql
          statement => "SELECT g.id AS id,g.product_name AS product_name,g.shop_id AS shop_id,g.category_id AS category_id,g.keyword AS keyword,g.status AS status FROM fg_product g WHERE g.updated_at > :sql_last_value"
          # 是否清除 last_run_metadata_path 的记录,如果为真那么每次都相当于从头开始查询所有的数据库记录
          clean_run =>false
          # 这是控制定时的，重复执行导入任务的时间间隔，第一位是分钟 不设置就是1分钟执行一次
          schedule => "* * * * *"
     }
}
output{
      elasticsearch{
                   # 要导入到的Elasticsearch所在的主机
                   hosts => "47.119.168.111:9200"
                   # 要导入到的Elasticsearch的索引的名称
                   index => "fob_index"
                   # 类型名称（类似数据库表名）
                   document_type => "fg_product"
                   # 主键名称（类似数据库表名）
                   document_id => "%{id}"
      }
      
      stdout{
            # JSON 格式输出
            codec => json_lines
      } 
}


查看mysql数据是否进入到es
k0CrP0

专题目录

ElasticStack-安装篇
ElasticStack-elasticsearch篇
ElasticStack-logstash篇
elasticSearch-mapping相关
elasticSearch-分词器介绍
elasticSearch-分词器实践笔记
elasticSearch-同义词分词器自定义实践
docker-elk集群实践
filebeat与logstash实践
filebeat之pipeline实践
Elasticsearch 7.x 白金级 破解实践
elk的告警调研与实践

发表于 更新于 阅读次数：

概述

主要汇总下基础的restful api

host:ip:9200

基础查询

获取es信息 GET host

request:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
{
    "name": "5f04c6c0a818",
    "cluster_name": "elasticsearch",
    "cluster_uuid": "vUPjZeMvSGqO_lrtmhqlmw",
    "version": {
        "number": "7.12.0",
        "build_flavor": "default",
        "build_type": "docker",
        "build_hash": "78722783c38caa25a70982b5b042074cde5d3b3a",
        "build_date": "2021-03-18T06:17:15.410153305Z",
        "build_snapshot": false,
        "lucene_version": "8.8.0",
        "minimum_wire_compatibility_version": "6.8.0",
        "minimum_index_compatibility_version": "6.0.0-beta1"
    },
    "tagline": "You Know, for Search"
}

获取索引 GET host/_cat/indices?v

request:

1
2
3
4
5
6
7
8
9
10
health status index                           uuid                   pri rep docs.count docs.deleted store.size pri.store.size
yellow open   bank                            ktIRbx9ZTwyDrEHlxiZpqw   1   1       1000            0    379.3kb        379.3kb
green  open   .kibana_task_manager_7.12.0_001 Yi3dSuiVSYWYaFidZxHSJg   1   0          9        26006      2.6mb          2.6mb
green  open   .apm-custom-link                DLOQIooBRiWD4O237c8tBA   1   0          0            0       208b           208b
green  open   .apm-agent-configuration        zUC91G_oRw63OY54rV2orw   1   0          0            0       208b           208b
green  open   .async-search                   NkCazI4YQDirjCrnvlEv2Q   1   0          0           24    794.5kb        794.5kb
green  open   .kibana_7.12.0_001              z4vax_yNTq2cOP7JxBphOQ   1   0         63           10      2.1mb          2.1mb
green  open   .kibana-event-log-7.12.0-000001 CYz7KhrsRa-dino1NEPDog   1   0          7            0     32.9kb         32.9kb
green  open   .tasks                          6wSie0zSRIW17VNQuxHgNQ   1   0          8            0     42.4kb         42.4kb

批量创建数据 POST host/bank/account/_bulk

request:

1
2
3
4
5
6
{"index":{"_id":"1"}}
{"account_number":1,"balance":39225,"firstname":"Amber","lastname":"Duke","age":32,"gender":"M","address":"880 Holmes Lane","employer":"Pyrami","email":"amberduke@pyrami.com","city":"Brogan","state":"IL"}
{"index":{"_id":"6"}}
{"account_number":6,"balance":5686,"firstname":"Hattie","lastname":"Bond","age":36,"gender":"M","address":"671 Bristol Street","employer":"Netagy","email":"hattiebond@netagy.com","city":"Dante","state":"TN"}
{"index":{"_id":"13"}}
{"account_number":13,"balance":32838,"firstname":"Nanette","lastname":"Bates","age":28,"gender":"F","address":"789 Madison Street","employer":"Quility","email":"nanettebates@quility.com","city":"Nogal","state":"VA"}

GET查询 GET host/bank/_search?q=age:31

response:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
{
    "took": 1,
    "timed_out": false,
    "_shards": {
        "total": 1,
        "successful": 1,
        "skipped": 0,
        "failed": 0
    },
    "hits": {
        "total": {
            "value": 61,
            "relation": "eq"
        },
        "max_score": 1.0,
        "hits": [
            {
                "_index": "bank",
                "_type": "account",
                "_id": "51",
                "_score": 1.0,
                "_source": {
                    "account_number": 51,
                    "balance": 14097,
                    "firstname": "Burton",
                    "lastname": "Meyers",
                    "age": 31,
                    "gender": "F",
                    "address": "334 River Street",
                    "employer": "Bezal",
                    "email": "burtonmeyers@bezal.com",
                    "city": "Jacksonburg",
                    "state": "MO"
                }
            }
        ]
    }
}

POST查询 POST host/bank/_search

request:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
{
    "query": {
        "bool": {
            "must": {
                "match_all": {}
            },
            "filter": {
                "range": {
                    "balance": {
                        "gte": 20000,
                        "lte": 30000
                    }
                }
            }
        }
    },
    "sort": {
        "age": {
            "order": "asc"
        }
    },
    "_source": [
        "account_number",
        "balance",
        "address"
    ],
    "from": 1,
    "size": 10
}

response:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
{
    "took": 1,
    "timed_out": false,
    "_shards": {
        "total": 1,
        "successful": 1,
        "skipped": 0,
        "failed": 0
    },
    "hits": {
        "total": {
            "value": 217,
            "relation": "eq"
        },
        "max_score": null,
        "hits": [
            {
                "_index": "bank",
                "_type": "account",
                "_id": "292",
                "_score": null,
                "_source": {
                    "account_number": 292,
                    "address": "691 Nassau Street",
                    "balance": 26679
                },
                "sort": [
                    20
                ]
            }
        ]
    }
}

聚合查询

分组 tags 后再分组source_ip

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
GET vpn-log-*/_search
{
  "query": {
    "bool": {
      "must": {
        "match": {
          "_index": "<vpn-log-{now/d{YYYY-MM-dd}}>"
        }
      },
      "filter": [
        {
          "terms": {
            "tags": [
              "QN"
            ]
          }
        }
      ]
    }
  },
  "aggs": {
    "topn": {
      "terms": {
        "field": "tags"
      },
      "aggs": {
        "source_ip_topn": {
          "terms": {
            "field": "source_ip"
          }
        }
      }
    }
  }
}

专题目录

ElasticStack-安装篇
ElasticStack-elasticsearch篇
ElasticStack-logstash篇
elasticSearch-mapping相关
elasticSearch-分词器介绍
elasticSearch-分词器实践笔记
elasticSearch-同义词分词器自定义实践
docker-elk集群实践
filebeat与logstash实践
filebeat之pipeline实践
Elasticsearch 7.x 白金级 破解实践
elk的告警调研与实践

发表于 更新于 阅读次数：

背景

NWd1w3

需要使用elasticSearch进行商品搜索及广告统计，对Elastic Stack进行调研.
这用nodejs客户端进行演示记录

安装

由于Elastic stack的所有Component都要互通，
要先设定一个network让所有的Container吃同一个网路如下:

docker network create elastic_stack

es安装

docker安装

1
2
3
4
5
6
7
8
9
10
docker run --name elasticsearch -p 9200:9200 -p 9300:9300 \
-e ES_JAVA_OPS="-Xms256m -Xmx256m" \
-e "discovery.type=single-node" \
--network elastic_stack \
-d docker.elastic.co/elasticsearch/elasticsearch:7.12.0

// 把配置文件拿出来
docker cp elasticsearch:/usr/share/elasticsearch/config /data/dockers/es/config
docker cp elasticsearch:/usr/share/elasticsearch/data /data/dockers/es/data
docker cp elasticsearch:/usr/share/elasticsearch/plugins /data/dockers/es/plugins

chmod -R 777 /data/dockers/es

1
2
3
4
5
6
7
8
docker run --name elasticsearch -p 9200:9200 -p 9300:9300 \
-e ES_JAVA_OPS="-Xms256m -Xmx256m" \
-e "discovery.type=single-node" \
--network elastic_stack \
-v /data/dockers/es/config:/usr/share/elasticsearch/config \
-v /data/dockers/es/data:/usr/share/elasticsearch/data \
-v /data/dockers/es/plugins:/usr/share/elasticsearch/plugins \
-d docker.elastic.co/elasticsearch/elasticsearch:7.12.0

验证是否正常：http://localhost:9200/

Kibana安装

1
2
3
docker run --name kibana \
--network elastic_stack \
-p 5601:5601 -d docker.elastic.co/kibana/kibana:7.12.0

验证是否正常：http://localhost:5601/

logstash安装

1. 首先创建一个容器 用来获取它的配置文件

1
2
docker run --name logstash \
-d docker.elastic.co/logstash/logstash:7.12.0

查看日志信息 是否启动成功

1
docker logs -f logstash

2. 创建挂载文件

拷贝数据

1
2
3
4
5
docker cp logstash:/usr/share/logstash/config /data/dockers/logstash/config

docker cp logstash:/usr/share/logstash/data /data/dockers/logstash/data

docker cp logstash:/usr/share/logstash/pipeline /data/dockers/logstash/pipeline

创建logstash配置文件路径

1
mkdir -p /data/dockers/logstash/config/conf.d

修改配置
logstash.yml

1
2
3
http.host: "0.0.0.0"
xpack.monitoring.elasticsearch.hosts: [ "http://elasticsearch:9200" ]
path.config: /usr/share/logstash/config/conf.d/*.conf  //配置mysql同步es配置使用
1
chmod -R 777 /data/dockers/logstash/

重新创建新的容器

1
2
3
4
5
6
7
8
9
10
11
docker run \
--name logstash \
--restart=always \
--network elastic_stack \
-p 5044:5044 \
-p 9600:9600 \
-e ES_JAVA_OPTS="-Duser.timezone=Asia/Shanghai" \
-v /data/dockers/logstash/config:/usr/share/logstash/config \
-v /data/dockers/logstash/data:/usr/share/logstash/data \
-v /data/dockers/logstash/pipeline:/usr/share/logstash/pipeline \
-d logstash:7.12.0

详细配置见ElasticStack-logstash篇

专题目录

ElasticStack-安装篇
ElasticStack-elasticsearch篇
ElasticStack-logstash篇
elasticSearch-mapping相关
elasticSearch-分词器介绍
elasticSearch-分词器实践笔记
elasticSearch-同义词分词器自定义实践
docker-elk集群实践
filebeat与logstash实践
filebeat之pipeline实践
Elasticsearch 7.x 白金级 破解实践
elk的告警调研与实践

发表于 更新于 阅读次数：

背景

为什么多阶段构建

  • Docker镜像是分层的，Dockerfile中的每个指令都会创建一个新的镜像层，镜像层可以被复用和缓存。当Dockerfile的指令修改了，复制的文件变化了，或者构建镜像时指定的变量不同了，对应的镜像层缓存就会失效，某一层的镜像缓存失效之后，它之后的镜像层缓存都会失效。

  • 因此我们还可以将RUN指令合并，但是需要记住的是，我们只能将变化频率一致的指令合并。

  • 我们应该把变化最少的部分放在Dockerfile的前面，这样可以充分利用镜像缓存。

  • 通过最小化镜像层的数量，我们可以得到更小的镜像。

实践

因为最近都在看nestjs,所以直接以nestjs来构建镜像

1
2
yarn global add @nestjs/cli
nest new nestjs-admin

稍等一会你会得到这样一个目录

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
./nestjs-admin
├── .gitignore
├── .prettierrc
├── README.md
├── nest-cli.json
├── package.json
├── src
│   ├── app.controller.spec.ts
│   ├── app.controller.ts
│   ├── app.module.ts
│   ├── app.service.ts
│   └── main.ts
├── test
│   ├── app.e2e-spec.ts
│   └── jest-e2e.json
├── tsconfig.build.json
├── tsconfig.json
├── tslint.json
└── yarn.lock

我们在根目录下创建一个 .dockerignore 文件，内容如下

.dockerignore 

1
2
3
4
5
node_modules  
.git  
.idea  
.vscode  
/coverage

然后我们在根目录继续创建一个 Dockerfile 文件，内容如下

Dockerfile

为了避免镜像中打包冗余的文件，我们使用多阶段构建镜像

大幅减小镜像体积的最简单和最快的方法是选择一个小得多的基本镜像。Alpine是一个很小的Linux发行版，可以完成这项工作。只要选择Node.js的Alpine版本，就会有很大的改进。

1
2
3
4
5
6
7
8
9
10
11
FROM node:12-alpine AS dependencies  
WORKDIR /usr/src/app  
COPY package.json yarn.lock ./  
RUN yarn install --production  
  
FROM node:12-alpine  
WORKDIR /usr/src/app  
COPY package.json dist ./  
COPY --from=dependencies /usr/src/app/node_modules ./node_modules  
EXPOSE 3000  
CMD [ "node", "dist/main" ]

package增加命令

1
"build:docker": "yarn build && docker build -t bulolo/$npm_package_name:latest . && docker push bulolo/$npm_package_name:latest",

运行docker打包镜像
yarn build:docker

发表于 更新于 阅读次数：

背景

最近在做一套RBAC的权限管理系统，当在设计部门及查询的时候，需要维护一个部门结构。结构包含 部门id,上级部门pid,当查询一个部门的所有子级部门时候，需要根据pid进行递归查询，层级越多，查询次数越多。那么怎么通过一个简单的数据库设计，满足查询某个部门的所有子级部门。

QAAAig

技巧

对于上图的部门层级结构，给出对应的数据库设计

4tKPL0

原理：
添加一个辅助的varchar字段pids,字段的逻辑是多个部门的id使用,来连接，假设首层使用0表示，每一个层级使用上一层的pids拼接上,再拼接上级部门id来表示，我们也可以通俗的理解为这个是部门层级pids为该部门的所有上级路径

举例：
技术部(id:1)的上级部门是长沙分公司(id:2),
长沙分公司的上级是集团(id:5)，
所以技术部的pids为:0,5,2,
长沙分公司的pids为:0,5

因此要查询长沙分公司的子部门，只需要用 0,5,2%去查询即可，即找到当前部门记录的pids(0,5), 拼上,再拼上当前部门ID(2)，再拼个%做后缀模糊匹配

1
pids like {pids},{id}%

发表于 更新于 阅读次数：

命令创建项目结构

module、controller、service创建

--no-spec不创建测试用例

1
2
3
nest g mo user
nest g controller user
nest g service user --no-spec

typeorm-model-generator

数据库表自动生成实体类

在package.json的scripts配置数据库生成实体类命令：

1
"db": "rm -rf src/entities & npx typeorm-model-generator -h localhost -d race -p 3306 -u root -x hong -e mysql -o src/entities --noConfig --relationIds --lazy --ce none --cp none"

npx typeorm-model-generator –help 查看帮助

1
2
3
4
5
6
7
8
9
10
11
12
13
14

-h dbhost
-d dbname
-p port
-u username
-x password
-e dbtype

-a 生成实体类 extends BaseEntity 可使用基础的CURD
-o 默认输出项目根目录下 output 文件夹下 可以自定义输出目标目录，例：-o ./src/demo/entities
--cf，将文件名转换为指定的大小写 [可选值："pascal"，"camel"，"none"] [默认值 none"] 
--ce pascal 生成实体类名采用人名式格式 [可选值："pascal"，"camel"，"none"] [默认值 none"] 
--cp camel 生成实体类属性采用驼峰 [可选值："pascal"，"camel"，"none"] [默认值 none”] 
--noConfig 不产生相关配置文件

使用方法：
1、将数据库所有表转成实体类

1
npm run db

2.单独将数据库的sys_user表生成实体类

1
npm run db -- --tables sys_user,sys_role

装饰器

装饰器 介绍
@Req() 获取Request对象(express)
@Res() 获取Response对象(express)
@Session() 获取req.session
@Param(key?: string) req.params / req.params[key],param是路由中的参数
@Body(key?: string) req.body / req.body[key]
@Query(key?: string) req.query / req.query[key],query是?后的参数
@Headers(name?: string) req.headers / req.headers[name]
@HttpCode(201) 指定response中的返回值

Modules

属性 介绍
providers 被Nest injector实例化的providers
controllers 被Nest injector实例化的controllers
imports 引入这些moludes export 的providers
exports providers的子集，被其他模块可以import

请求生命周期

周期 xm1jf3

顺序：

  • 中间件
  • 卫兵
  • 拦截器（在操纵流之前）
  • 管道
  • 服务端路由函数
  • 拦截器（在操纵流之后）
  • 异常过滤器（如果捕获到任何异常）

中间件

请求生命周期的第一个流程是经过中间件，我们来实现一个日志中间件

如何实现

中间件必须实现NestMiddleware

./common/middleware/logger.middleware.ts

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
import { Injectable, NestMiddleware } from '@nestjs/common';
import { Request, Response } from 'express';

@Injectable()
export class LoggerMiddleware implements NestMiddleware {
  use(req: Request, res: Response, next: Function) {
    const logFormat = `  >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
      Request original url: ${req.originalUrl}
      Method: ${req.method}
      IP: ${req.ip}
      Status code: ${res.statusCode}
      Parmas: ${JSON.stringify(req.params)}
      Query: ${JSON.stringify(req.query)}
      Body: ${JSON.stringify(req.body)} \n  >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> 
    `
    console.log(logFormat);
    next();
  }
}

如何注册中间件

  • 全局中间件
    在main.ts中进行全局注册
1
2
import { LoggerMiddleware } from './common/middleware/logger.middleware'
app.use(new LoggerMiddleware().use)

发表于 更新于 阅读次数：

消费模式

独占订阅(流)：

顾名思义，在任何给定时间内，订阅（消费者组）中只有一个消费者消费主题分区。 下面的图1说明了独占订阅的示例。 有一个有订阅A的活动消费者A-0消息m0到m4按顺序传送并由A-0消费。 如果另一个消费者A-1想要附加到订阅A，则不允许这样做。
B3MuuA

独占订阅

故障转移订阅(Failover sub streaming):使用故障转移订阅，多个使用者可以附加到同一订阅。 但是，对于给定的主题分区，将选择一个使用者作为该主题分区的主使用者，其他消费者将被指定为故障转移消费者，当主消费者断开连接时，分区将被重新分配给其中一个故障转移消费者，而新分配的消费者将成为新的主消费者。 发生这种情况时，所有未确认的消息都将传递给新的主消费者，这类似于Apache Kafka中的使用者分区重新平衡。 图2显示了故障转移订阅，消费者B-0和B-1通过订阅B订阅消费消息.B-0是主消费者并接收所有消息，B-1是故障转移消费者，如果消费者B-0出现故障，将接管消费。
oq0Quo

故障转移订阅

共享订阅(队列)：使用共享订阅，可以将所需数量的消费者附加到同一订阅。消息以多个消费者的循环尝试分发形式传递，并且任何给定的消息仅传递给一个消费者。当消费者断开连接时，所有传递给它并且未被确认的消息将被重新安排，以便发送给该订阅上剩余的剩余消费者。图3说明了共享订阅。消费者C-1，C-2和C-3都在同一主题分区上消费消息。每个消费者接收大约1/3的消息。如果您想提高消费率，您可以在不增加分区数量的情况下为更多的消费者提供相同的订阅（尽可能多的消费者）。
cT304g

共享订阅

独占和故障转移订阅仅允许每个订阅每个主题分区仅有一个消费者。它们按分区顺序使用消息。它们最适用于需要严格排序的流用例。另一方面，共享订阅允许每个主题分区有多个消费者，同一订阅中的每个消费者仅接收发布到主题分区的一部分消息。共享订阅最适用于不需要排序的并且可以扩展超出分区数量的使用者数量的队列用例。

Pulsar中的subscription(订阅)实际上与Apache Kafka中的消费者群体相同。创建订阅具有高度可扩展性且非常低廉的。可以根据需要创建任意数量的订阅，对同一主题的不同订阅不必具有相同的订阅类型。这意味着可以在同一主题上有10个消费者的故障转移订阅或有20个消费者的共享订阅。如果共享订阅处理事件的速度很慢，则可以在不更改分区数的情况下向共享订阅添加更多消费者。图4描绘了一个包含3个订阅A，B和C的主题，并说明了消息如何从生产者流向消费者。
rRtalN

除了统一消息传递API之外，由于Pulsar主题分区实际上是存储在Apache BookKeeper中的分布式日志，它还提供了一个读取器(reader) API（类似于消费者(consumer) API但没有游标管理），以便用户完全控制如何使用消息本身。

消息确认(Message Ackmowledgment)

当使用跨机器分布的消息传递系统时，可能会发生故障。在消费者从消息传递系统中的主题消费消息的情况下，消费消息的消费者和服务于主题分区的消息代理都可能失败。当发生这样的故障时，能够从消费者停止的地方恢复消费，这样既不会错过消息，也不必处理已经确认的消息。在Apache Kafka中，恢复点通常称为偏移，更新恢复点的过程称为消息确认或提交偏移。在Apache Pulsar中，游标(cursors)用于跟踪每个订阅(subscription)的消息确认(message acknowledgment)。每当消费者在主题分区上确认消息时，游标都会更新，更新游标可确保消费者不会再次收到消息，但是游标并不像Apache Kafka那样简单。Apache Pulsar有两种方法可以确认消息，个体确认ack或累积确认消息。通过累积确认，消费者只需要确认它收到的最后一条消息，主题分区中的所有消息（包括）提供消息ID将被标记为已确认，并且不会再次传递给消费者，累积确认与Apache Kafka中的偏移更新实际上相同。Apache Pulsar的区别特征是能够个体单独进行ack，也就是选择性acking。消费者可以单体确认消息。Acked消息将不会被重新传递。图5说明了ack个体和ack累积之间的差异（灰色框中的消息被确认并且不会被重新传递）。在图的顶部，它显示了ack累积的一个例子，M12之前的消息被标记为acked。在图的底部，它显示了单独进行acking的示例。仅确认消息M7和M12 - 在消费者失败的情况下，除了M7和M12之外，将重新传送所有消息。
NIpfft

独占(exclusive)或故障转移(failover)订阅的消费者能够单个或累积地发送消息(ack message);而共享订阅中的消费者只允许单独发送消息(ack messages)。单独确认消息的能力为处理消费者故障提供了更好的体验。对于某些应用来说，处理那些已经确认过的消息可能是非常耗时的，防止重新传送已经确认的消息是非常重要。

发表于 更新于 阅读次数：

关于

最近，俊瑶先生一直在研究 Pulsar.

俊瑶先生是一名 忠实的Kafka 的用户，为kafka填过几次文章，如：2018-12-01:kafka消息列队 ，同时也将kafka运用于多个项目实践中，上一个项目为一个招投标的订阅消息推送。俊瑶先生于今年1月份透过涂鸦iot平台关注到了Pulsar，当时很疑惑为什么涂鸦选择了Pulsar，于是开展了长达3个月断断续续调研。期间不乏将外网所有Pulsar的文章都看了一遍，许多pulsar文章都与kafka进行了深入的对比
otIEmm

为什么Pulsar

G8gwE2 构建消息基础设施的第一步是选择合适的消息中间件技术。在这方面有很多选择，从各种开源框架（如 RabbitMQ、ActiveMQ、NATS）到一些商用产品（如 IBM MQ 或者 RedHat AMQ）。当然，除了这些之外，我们还有 Kafka。不过，我最后并没有选择使用已久的Kafka，而是选择了 Pulsar。

与Kafka对比

WLgzG8

模型概念

  • Kafka： Producer - topic - consumer group - consumer
  • Pulsar： Producer - topic - subscription - consumer

消费模式

  • Kafka： 主要集中在流（Stream）模式，对单个 partition 是独占消费，没有共享（Queue）的消费模式；
  • Pulsar：提供了统一的消息模型和 API。流（Stream）模式——独占和故障切换订阅方式；队列（Queue）模式——共享订阅的方式。

消息确认（Ack）

  • Kafka： 使用偏移 Offset；
  • Pulsar：使用专门的 Cursor 管理。累积确认和 Kafka 效果一样；提供单条或选择性确认。

消息保留

  • Kafka：根据设置的保留期来删除消息。有可能消息没被消费，过期后被删除。 不支持 TTL。
  • Pulsar：消息只有被所有订阅消费后才会删除，不会丢失数据。也允许设置保留期，保留被消费的数据。支持 TTL。

详细的消费模式和消息确认请参考：pulsar的消费模式

Pulsar的系统架构和设计理念

Pulsar 的分层架构

xgcpzq

Broker层（或无状态服务层）架构

j88ypw

架构上 Pulsar 和 Kafka 的对比

Apache Kafka 是以分区为存储中心，而 Apache Pulsar 是以 Segment 为存储中心。如下图：
mHIByg

  • 在 Apache Kafka 中，分区只能存储在单个节点上并复制到其他节点，其容量受最小节点容量的限制。这意味着容量扩展需要对分区重新平衡，这反过来又需要重新复制整个分区，以平衡新添加的代理的数据和流量。重新传输数据非常昂贵且容易出错，并且会消耗网络带宽和 I/O。维护人员在执行此操作时必须非常小心，以避免破坏生产系统。
  • Kafka 中分区数据的重新拷贝不仅发生在以分区为中心的系统中的群集扩展上。许多其他事情也会触发数据重新拷贝，例如副本故障，磁盘故障或计算机的故障。在数据重新复制期间，分区通常不可用，直到数据重新复制完成。例如，如果您将分区配置为存储为 3 个副本，这时，如果丢失了一个副本，则必须重新复制完整个分区后，分区才可以再次可用。

Pulsar的无缝Broker故障恢复

下图说明了 Pulsar 如何处理 Broker 失败的示例。 在例子中 Broker 2 因某种原因（例如停电）而断开。 Pulsar 检测到 Broker 2 已关闭，并立即将 Topic1-Part2 的所有权从 Broker 2 转移到 Broker 3。在 Pulsar 中数据存储和数据服务分离，所以当代理 3 接管 Topic1-Part2 的所有权时，它不需要复制 Partiton 的数据。 如果有新数据到来，它立即附加并存储为 Topic1-Part2 中的 Segment x + 1。 Segment x + 1 被分发并存储在 Bookie1, 2 和 4 上。因为它不需要重新复制数据，所以所有权转移立即发生而不会牺牲主题分区的可用性。
cAIxTI

Pulsar的无缝存储（Bookie）故障恢复

下图说明了 Pulsar（通过 Apache BookKeeper）如何处理 bookie 的磁盘故障。 这里有一个磁盘故障导致存储在 bookie 2 上的 Segment 4 被破坏。Apache BookKeeper 后台会检测到这个错误并进行复制修复。
uFhIki
Apache BookKeeper 中的副本修复是 Segment（甚至是 Entry）级别的多对多快速修复，这比重新复制整个主题分区要精细，只会复制必须的数据。 这意味着 Apache BookKeeper 可以从 bookie 3 和 bookie 4 读取 Segment 4 中的消息，并在 bookie 1 处修复 Segment 4。所有的副本修复都在后台进行，对 Broker 和应用透明。
即使有 Bookie 节点出错的情况发生时，通过添加新的可用的 Bookie 来替换失败的 Bookie，所有 Broker 都可以继续接受写入，而不会牺牲主题分区的可用性。

经过调研及等待时机：终于来了一个场景是非常适合使用pulsar

本篇将记录下pulsar使用过程遇到的问题及过程。

安装篇

首先我需要有一个mac本地环境进行pulsar进行调试，以及一个线上的linux正式环境进行部署上线.同时俊瑶先生本次使用的nodejs的pulsar版本。
根据官方文档 pulsar安装文档,要使用pulsar的nodejs版本，需要安装Pulsar C++库.

安装 C++库请阅读文档 Pulsar C++ client

mac安装

mac安装的时候，务必安装xcode,如果是beta版本的，务必指定选择xcode的beta，从而达到装mac C++的要求,也即gcc
然后通过brew安装libpulsar
brew install libpulsar
全局安装 pulsar-client
npm install pulsar-client -g

linux安装

linux服务器采用ubuntu,下载Debian package
并通过 apt install ./apache-pulsar-client*.deb进行安装

client

client-devel

实践使用

TODO:

发表于 更新于 阅读次数：

经常操作数组，统一做一次整理吧

entries、keys、values

示例

ES6 提供三个新的方法——entries()，keys()和values()——用于遍历数组。

keys

1
2
3
4
5
for (let index of ['a', 'b'].keys()) {
  console.log(index);
}
// 0
// 1

values

1
2
3
4
5
for (let item of ['a', 'b'].values()) {
  console.log(item);
}
// 'a'
// 'b'

entries

1
2
3
4
for (let [index, value] of ['a', 'b'].entries()) {
  console.log(index, value);
}
// 0 "a"

fromEntries

一般结合Object.fromEntries数组转对象

1
2
3
4
5
6
7
8
9
10
11
12
13
14
// 对象转数组
const obj = {
    a: '1',
    b: 2
}
const arr = Object.entries(obj)
console.log(arr) // [ [ 'a', '1' ], [ 'b', 2 ] ]

// 数组转对象
const obj = Object.fromEntries([
    ['a', '1'],
    ['b', 2]
])
console.log(obj) //{ a: '1', b: 2 }

concat

合并多个数组，返回合并后的新数组，原数组没有变化。

1
2
const array = [1,2].concat(['a', 'b'], ['name']);
// [1, 2, "a", "b", "name"]

另外也可以通过

1
2
3
const arr1 = [1, 2]
const arr2 = ['a', 'b']
const array = [...arr2, arr2]

indexOf

indexOf()方法返回在该数组中第一个找到的元素位置，如果它不存在则返回-1。

语法

1
array.indexOf(item,start)
  • item 必须。查找的元素。
  • start 可选的整数参数。规定在数组中开始检索的位置。它的合法取值是 0 到 stringObject.length - 1。如省略该参数，则将从字符串的首字符开始检索。

示例

1
2
var arr = ['apple','orange','pear'];
console.log("found:", arr.indexOf("orange") != -1);

forEach

只要是for循环能够遍历的数组，forEach都可以实现。

语法

1
array.forEach(function(currentValue, index, arr), thisValue)
  • currentValue 必需。当前元素
  • index 可选。当前元素的索引值。
  • arr 可选。当前元素所属的数组对象。

示例

1
2
3
4
5
6
7
8
let arr1 = [1, 5, 7, 8, 9, 6]
let str = ''

arr1.forEach((item, index, array) => {
  console.log(item, index, array)
  str += `<li>${item}</li>`
})
console.log(str)

some every

  • some方法 如果数组中有一个元素满足条件，返回true 否则返回false
  • every方法 如果数组中所有元素都满足条件则返回true 否则返回false

语法

1
2
array.some(function(currentValue,index,arr),thisValue)
array.every(function(currentValue,index,arr), thisValue)
  • currentValue 必须。当前元素的值
  • index 可选。当前元素的索引值
  • arr 可选。当前元素属于的数组对象

示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
let arr = [5, 7, 8, 9, 6, ]
const v = arr.some(item => {
    return item < 6
})
const x = arr.every(item => {
    return item < 9
})

console.log(v) //true
console.log(x) //false

const obj = [{
        name: '小明',
        age: 9,
        sex: '男'
    },
    {
        name: '小红',
        age: 10,
        sex: '女'
    }
]
const a = obj.some(item => {
    return item.name == '小红'
})
const b = obj.every(item => {
    return item.age == 9
})
console.log(a) //true
console.log(b) //false

filter

  • 将数组中符合条件的元素留下来组成一个新的数组
  • filter() 不会改变原始数组。

语法

1
array.filter(function(currentValue,index,arr), thisValue)
  • currentValue 必须。当前元素的值
  • index 可选。当前元素的索引值
  • arr 可选。当前元素属于的数组对象

示例

filter简单过滤

1
2
3
4
5
let arr = [1, 5, 6, 3, 4, 2, 8, 7];
const a = arr.filter(item => {
    return item < 5
});
console.log(a) //[1,3,4,2]

filter数组去重

1
2
3
4
5
6
7
8
9
10
var array = [1, 2, 1, 2, 4, 5, 3, 5, 6, 7, 6];
// 检测数组下标，是否等于当前项的下标，不相等则认为有重复
var result = array.filter((item, index, arr) => {
    // item : 数组每一项的值
    // index: 每一项的下标
    // arr: 原始数组
    return index === arr.indexOf(item);
})

console.log(result); //[1, 2, 4, 5,3, 6, 7]

find

  • 遍历数组，当找到第一个符合条件的元素的时候，就会停止遍历，并返回这个元素，如果没找到符合条件的数组就输出undefined
    实例：

语法

1
array.find(function(currentValue, index, arr),thisValue)
  • currentValue 必需。当前元素
  • index 可选。当前元素的索引值
  • arr 可选。当前元素所属的数组对象

示例

1
2
3
4
5
6
7
8
9
let arr = [1, 5, 6, 3, 4, 2, 8, 7];
  const b = arr.find(item => {
      return item > 9
  });
  const c = arr.find(item => {
      return item < 9
  });
  console.log(b) //undefined
  console.log(c) //1

includes

  • 只要对应数组中包含这个元素，就返回true, 否则返回false

语法

1
2
3
arr.includes(searchElement)
arr.includes(searchElement, fromIndex)

  • searchElement必须。需要查找的元素值。
  • fromIndex可选。从该索引处开始查找 searchElement。如果为负值，则按升序从 array.length + fromIndex 的索引开始搜索。默认为 0。

示例：

1
2
3
4
5
let arr = [1, 5, 6, 3, 4, 2, 8, 7];
const d = arr.includes(7)
const x = arr.includes(9)
console.log(d) //true
console.log(x) //false

map

  • 根据已知数组把已知数组作为元素生成新的的数组，新数组的长度与已知数组长度一致

语法

1
array.map(function(currentValue,index,arr), thisValue)
  • currentValue 必须。当前元素的值
  • index 可选。当前元素的索引值
  • arr 可选。当前元素属于的数组对象

示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
let arr = [1, 5, 6, 3, 4, 2, 8, 7];
 const y = arr.map((item, index, array) => {
     return item, array
 })
 console.log(y)
 //   (8)[Array(8), Array(8), Array(8), Array(8), Array(8), Array(8), Array(8), Array(8)]
 //   0: (8)[1, 5, 6, 3, 4, 2, 8, 7]
 //   1: (8)[1, 5, 6, 3, 4, 2, 8, 7]
 //   2: (8)[1, 5, 6, 3, 4, 2, 8, 7]
 //   3: (8)[1, 5, 6, 3, 4, 2, 8, 7]
 //   4: (8)[1, 5, 6, 3, 4, 2, 8, 7]
 //   5: (8)[1, 5, 6, 3, 4, 2, 8, 7]
 //   6: (8)[1, 5, 6, 3, 4, 2, 8, 7]
 //   7: (8)[1, 5, 6, 3, 4, 2, 8, 7]

reduce

定义和用法

  • reduce() 方法接收一个函数作为累加器，数组中的每个值（从左到右）开始缩减，最终计算为一个值。
  • reduce() 可以作为一个高阶函数，用于函数的 compose。
    注意: reduce() 对于空数组是不会执行回调函数的。

语法

1
array.reduce(function(total, currentValue, currentIndex, arr), initialValue)
  • total 必需。初始值, 或者计算结束后的返回值。
  • currentValue 必需。当前元素
  • currentIndex 可选。当前元素的索引
  • arr 可选。当前元素所属的数组对象。

示例

reduce累加求和

1
2
3
4
5
6
const arr = [3,9,4,3,6,0,9]
const sum = arr.reduce((pre, item) => {
    return pre + item
}, 0)
console.log(sum) // 34

reduce数组最大值

1
2
3
4
5
const arr = [3,9,4,3,6,0,9]
var max = arr.reduce((prev, cur) => {
    return Math.max(prev,cur)
})
console.log(max) // 9

reduce数组去重

1
2
3
4
5
6
const arr = [3,9,4,3,6,0,9]
var newArr = arr.reduce((prev, cur) => {
    prev.indexOf(cur) === -1 && prev.push(cur);
    return prev;
},[]);
console.log(newArr) // [ 3, 9, 4, 6, 0 ]

reduce数组对象去重

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
let person = [
     {id: 0, name: "小明"},
     {id: 1, name: "小张"},
     {id: 2, name: "小李"},
     {id: 3, name: "小孙"},
     {id: 1, name: "小周"},
     {id: 2, name: "小陈"},  
];
 
let obj = {};
 
let peon = person.reduce((cur,next) => {
    obj[next.id] ? "" : obj[next.id] = true && cur.push(next);
    return cur;
},[]) //设置cur默认类型为数组，并且初始值为空的数组
console.log(peon);

sort

语法

1
array.sort(sortfunction)

示例

1
2
3
4
var points = [40, 100, 1, 5, 25, 10]
points.sort( (a, b) => {
    return a - b
})

Set

示例

Set数组去重

1
2
3
4
const arr = [3,9,4,3,6,0,9]
let set = new Set(arr)
const newArr = Array.from(set)
console.log(newArr) // [ 3, 9, 4, 6, 0 ]
0%