本文介绍下Elasticsearch集群的相关知识。

什么是Elasticsearch

Elasticsearch 是一个分布式、高扩展、高实时的搜索与数据分析引擎。它能很方便的使大量数据具有搜索、分析和探索的能力。充分利用Elasticsearch的水平伸缩性，能使数据在生产环境变得更有价值。

为什么要用集群

用ElasticSearch集群，将单个索引的分片到多个不同分布式物理机器上存储，从而实现高可用、容错性等。ElasticSearch会自动选举实现主备服务器。

达到基本高可用，一般要至少启动3个节点，3个节点互相连接，单个节点包括所有角色，其中任意节点停机集群依然可用。

关键概念

节点角色

• Master，集群管理
• Voting，投票选举节点
• Data，数据节点
• Ingest，数据编辑节点
• Coordinate，协调节点
• Machine Learning，集群学习节点

索引

一般意义上的索引是一种基于文档（数据）生成、建立的，用于快速定位指定文档的工具。而 ElasticSearch 对索引的定义有所不同，ElasticSearch 中的索引对应 MySQL 中的 Database ，也就说 ElasticSearch 中的索引更像是一种数据存储集合，即用于存储文档。

分片

分片（Shard），是Elasticsearch中的最小存储单元。

• Shards分片：代表索引分片，ElasticSearch可以把一个完整的索引分成多个分片，这样的好处是可以把一个
  大的索引拆分成多个，分布到不同的节点上。构成分布式搜索。分片的数量只能在索引创建前指定，并且索引
  创建后不能更改。
• Replicas分片：代表副本分片，ElasticSearch可以设置多个索引的副本，副本的作用一是提高系统的容错性，
  当某个节点某个分片损坏或丢失时可以从副本中恢复。二是提高ElasticSearch的查询效率，ElasticSearch
  会自动对搜索请求进行负载均衡。

数据路由

• 当客户端发起创建document的时候，ElasticSearch需要确定这个document放在该索引哪个shard上。这个过程就是数据路由。
• 路由算法：分片位置shard=hash(routing) % number_of_primary_shards
• 如果number_of_primary_shards在查询的时候取余发生变化，则无法获取到该数据。即：在查询的时候，底层根据文档id%主分片数量获取分片位置

1
2
3

1.客户端向 ES1节点（协调节点）发送写请求，通过路由计算公式得到值为0，则当前数据应被写到主分片 S0 上。
2.ES1 节点将请求转发到 S0 主分片所在的节点 ES3，ES3 接受请求并写入到磁盘。
3.并发将数据复制到两个副本分片 R0 上，其中通过乐观并发控制数据的冲突。一旦所有的副本分片都报告成功，则节点 ES3 将向协调节点报告成功，协调节点向客户端报告成功。

脑裂现象

正常情况下，集群中的所有节点，应该对主节点的选择是一致的，即一个集群中只有一个选举出来的主节点。然而，在某些情况下，比如网络通信出现问题、主节点因为负载过大停止响应等等，就会导致重新选举主节点，此时可能会出现集群中有多个主节点的现象，即节点对集群状态的认知不一致，称之为脑裂现象。

为了尽量避免此种情况的出现，可以通过discovery.zen.minimum_master_nodes来设置最少可工作的候选主节点个数，建议设置为(候选主节点数 / 2) + 1。保证集群中有半数以上的候选主节点。

安装

现在开始介绍如何通过docker搭建一个简单的Elasticsearch集群服务

前面介绍了rbac组件在单点服务上的基本应用场景,今天来介绍一下，在微服务架构下如何使用rbac来控制路由权限。

服务架构

这是一个很普通的微服务架构，用户通过请求服务器的路由资源，发起访问，各服务需要检查用户是否拥有权限。

RBAC组件是将权限规则存入内存里，当管理员通过权限管理后台更新的权限信息后，如何让各服务节点即时加载最新的权限规则呢？

我们发现，管理后台产生新的权限规则，多个服务节点读取最新的权限规则，这是一个典型的生产者/消费者模型。

casbin有提供一个Watcher接口。官网也提供了如redis、etcd、kafka等解决访问，详细文档请参考：https://casbin.org/docs/en/watchers。下面就用redis做示例说明

redis

基于redis的发布/订阅(Publish/Subscribe),管理员更新用户权限后，通过redis发布一条信息，其余服务节点，因为订阅了该信息，故每个客户端都会接受到一条通知，此刻更新内存的权限规则，即可实现分布式的权限更新。将此组件称为:redis-watcher

• Demo

  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 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131

  package main import ( "fmt" "github.com/casbin/casbin/v2/persist" uuid "github.com/satori/go.uuid" "reflect" "sync" "github.com/go-redis/redis" ) // Watcher type Watcher struct { options options // 选项 pubConn redis.UniversalClient //生产消息 subConn redis.UniversalClient //订阅消息 callback func(string) // 回调方法 closed chan struct{} once sync.Once } // 选项 type options struct { id string // 当前节点的id channel string // redis的管道名称 } // Option 选项接口 type Option interface { Set(options *options) } // channel选项 type channelOption string func (o channelOption) Set(opts *options) { opts.channel = string(o) } func WithChannel(channel string) Option { return channelOption(channel) } // id选项 type idOption string func (o idOption) Set(opts *options) { opts.id = string(o) } func WithID(id string) Option { return idOption(id) } // NewWatcher 实例化 func NewWatcher(conn redis.UniversalClient, opts ...Option) (persist.Watcher, error) { w := &Watcher{ closed: make(chan struct{}), pubConn: conn, subConn: conn, } w.options = options{ channel: "/casbin", id : uuid.NewV4().String(), } for _, opt := range opts { opt.Set(&w.options) } go func() { for { select { case <-w.closed: return default: err := w.subscribe() if err != nil { fmt.Printf("Failure from Redis subscription: %v", err) } } } }() return w, nil } // SetUpdateCallBack sets the update callback function invoked by the watcher // when the policy is updated. Defaults to Enforcer.LoadPolicy() func (w *Watcher) SetUpdateCallback(callback func(string)) error { w.callback = callback return nil } // Update publishes a message to all other casbin instances telling them to // invoke their update callback func (w *Watcher) Update() error { // 当有权限更新时，发送当前节点的ID,通知其他节点有更新 if _, err := w.pubConn.Publish(w.options.channel, w.options.id).Result(); err != nil { return err } return nil } // Close disconnects the watcher from redis func (w *Watcher) Close() { w.once.Do(func() { close(w.closed) _ = w.subConn.Close() _ = w.pubConn.Close() }) } // subscribe 订阅redis func (w *Watcher) subscribe() error { psc := w.subConn.Subscribe(w.options.channel) defer psc.Unsubscribe() for { receive, err := psc.Receive() fmt.Println(reflect.TypeOf(receive)) if err != nil { fmt.Printf("try subscribe channel[%s] error[%s]\n", w.options.channel, err.Error()) return nil } switch n := receive.(type) { case error: return n case *redis.Message: // 需要跳过当前生产者 if n.Payload != w.options.id && w.callback != nil { w.callback("success") // 执行回调 } case *redis.Subscription: if n.Count == 0 { return nil } } } }

• 使用方式：

  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

  func main() { authEnforcer, err := casbin.NewEnforcerSafe("rbac/conf/rbac_model.conf") if err != nil { log.Fatal(err) } redisConn := redis.NewClient( &redis.Options{ Addr: "127.0.0.1", Password: "", DB: 0, }) if _, err := redisConn.Ping().Result(); err != nil { log.Fatal(err) } watcher, err := NewWatcher(redisConn) if err != nil { log.Fatal(err) } watcher.SetUpdateCallback(func(s string) { // 当有权限更新时，会执行这个回调，具体更新内存的权限逻辑可以自定义 enforcer.LoadPolicy() }) authEnforcer.SetWatcher(watcher) }

在实际的场景中，正常的请求不会造成系统异常。但我们也需要防止异常高并发场景导致整个系统崩溃的情况出现。下面介绍限流器的使用。

为什么需要限流器

用户增长过快、热门业务或者爬虫等恶意攻击行为致使请求量突然增大，比如学校的教务系统，到了查分之日，请求量涨到之前的 100 倍都不止，没多久该接口几乎不可使用，并引发连锁反应导致整个系统崩溃。

通过限流器，给接口设置最高访问量，保证在系统的可承受范围内，提供服务。一旦超过该阈值，不允许用户访问，保证系统的其他服务能继续正常运行。

实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

package limiter

import (
	"time"

	"golang.org/x/time/rate"
)

// Limiter 限流器
type Limiter struct {
	*rate.Limiter
}

// New 实例化
func New(maxRate int) *Limiter {
	instance := new(Limiter)
	instance.Limiter = rate.NewLimiter(rate.Every(time.Second), maxRate)

	return instance
}

示例

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

package main
// LimitMiddleware 限流器中间件
func LimitMiddleware(maxRate int) gin.HandlerFunc {
	limiter := New(maxRate) // 实例化一个限流器
	return func(ctx *gin.Context) {
		if !limiter.Allow() { // 这里用于控制当限流器达到阈值的判断
			ctx.JSON(403, gin.H{"message": "Busy"})
			ctx.Abort()
		}
		ctx.Next()
	}
}

func main() {
	r := gin.Default()
	r.Use(LimitMiddleware(100)) // 每秒最高允许100个请求
	r.GET("/test", func(c *gin.Context) {
		c.JSON(200, gin.H{
			"message": "pong",
		})
	})
	r.Run()
}

当并发达到100以上时，后面的请求都会返回403。这样可以保证不会给系统更大的压力。

原理

请参考：https://zhuanlan.zhihu.com/p/140440501

在高并发场景中，我们最普遍的方案是采用高性能的缓存。在缓存使用过程中，需要注意缓存穿透、缓存雪崩、缓存击穿。以下介绍如何通过singleflight防止缓存击穿。

什么是缓存击穿

缓存击穿是指缓存中没有但数据库中有的数据（一般是缓存时间到期），这时由于并发用户特别多，同时读缓存没读到数据，又同时去数据库去取数据，引起数据库压力瞬间增大，造成过大压力。

SingleFlight

在多个并发请求对一个失效的key进行源数据获取时，只让其中一个得到执行，其余阻塞等待到执行的那个请求完成后，将结果传递给阻塞的其他请求达到防止击穿的效果。

安装

1

go get golang.org/x/sync/singleflight

示例

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

package main

import (
	"fmt"
	"log"
	"time"

	"golang.org/x/sync/singleflight"
)

var singleHandle singleflight.Group

func main() {
	for i := 0; i < 10; i++ { // 模拟10个并发
		go func() {
			fmt.Println(GetName("username"))
		}()
	}
	// 等待协程执行结束
	time.Sleep(3 * time.Second)
}

// GetName 获取名称
func GetName(cacheKey string) string {
	result, _, _ := singleHandle.Do(cacheKey, func() (ret interface{}, err error) {
		log.Printf("getting %s from database\n", cacheKey)
		log.Printf("setting %s in cache\n", cacheKey)
		return "hongker", nil
	})
	return result.(string)
}

• 结果

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

  2021/02/18 14:26:12 getting username from database 2021/02/18 14:26:12 setting username in cache hongker hongker hongker hongker hongker hongker hongker hongker hongker hongker

  只有一个协程执行从数据库中获取并设置缓存的操作，其他协程则是在等待获取缓存。避免缓存击穿。

本篇文章主要介绍golang里集合的实现思路。

什么是集合？

集合通常是由一组无序的, 不能重复的元素构成

功能

• 添加元素
• 删除元素
• 获取集合的元素数量
• 检查集合是否包含某元素
• 清空集合
• 判断集合是否为空
• 集合转数组
• 复制

接口设计如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

type Set interface {
	// 添加元素
	Add(items ...interface{})
	// 包含
	Contain(item interface{}) bool
	// 删除
	Remove(item interface{})
	// 集合大小
	Size() int
	// 清空
	Clear()
	// 判断是否为空
	Empty() bool
	// 创建副本
	Duplicate() Set
	// 数组
	ToSlice() []interface{}
}

具体实现

本文介绍在golang里如何实现队列得数据结构

概念

队列是一种特殊的线性表，特殊之处在于它只允许在表的前端（front）进行删除操作，而在表的后端（rear）进行插入操作，和栈一样，队列是一种操作受限制的线性表。进行插入操作的端称为队尾，进行删除操作的端称为队头。队列中没有元素时，称为空队列。

特性

• 入列
  从尾部将元素加入到队列里。
• 出列
  从队列头部获取元素。
• 查看长度
  获取队列存储的元素个数。
• 清空队列
  将队列里的数据清空。

Demo

• 定义Queue接口

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

  // Queue 队列 type Queue interface { // Push 入列 Push(items ...interface{}) // Pop 出列 Pop() (interface{}) // MPop 批量出列 MPop(n int) ([]interface{}) // Length 长度 Length() int // Clear 清空 Clear() }

本文介绍如何巧妙的使用go-swagger自动生成API接口文档。

什么是swagger

Swagger 是一个规范和完整的框架，用于生成、描述、调用和可视化 RESTful 风格的 Web 服务。

准备工作

• 下载swag工具

  1 2	// 通过go get下载并编译，在$GOPATH/bin目录下会生成swag命令，用于扫描注解。 go get -u github.com/swaggo/swag/cmd/swag

• 引入ego框架,已集成好的gin-swagger组件

  1	go get github.com/ebar-go/ego

API文档初始化

• 先写项目说明

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

  // @title Swagger Example API // @version 1.0 // @description This is a sample server Petstore server. // @termsOfService http://swagger.io/terms/ // @contact.name hongker // @contact.url http://hongker.github.io // @contact.email xiaok2013@live.com // @license.name Apache 2.0 // @license.url http://www.apache.org/licenses/LICENSE-2.0.html // @host localhost:8080 // @BasePath /v1 func main() { s := http.NewServer() // 加载路由 route.Load(s.Router) // 启动 secure.Panic(s.Start()) }

本文介绍golang中RBAC(基于角色的权限控制)的使用方式

什么是RBAC

RBAC 是基于角色的访问控制（Role-Based Access Control ）在 RBAC 中，权限与角色相关联，用户通过成为适当角色的成员而得到这些角色的权限。这就极大地简化了权限的管理。这样管理都是层级相互依赖的，权限赋予给角色，而把角色又赋予用户，这样的权限设计很清楚，管理起来很方便。

实现

我选用的是github.com/casbin/casbin,选择原因:社区的认同(Star很高)。

安装

1

go get github.com/casbin/casbin

定义模型

• conf/rbac_model.conf

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

  [request_definition] r = sub, obj, act [policy_definition] p = sub, obj, act [role_definition] g = _, _ [policy_effect] e = some(where (p.eft == allow)) [matchers] m = r.sub == p.sub && keyMatch(r.obj, p.obj) && (r.act == p.act || p.act == "*")

• conf/policy.csv
  权限细则，可以读csv文件，可以读数据库。。

  1 2 3 4 5 6 7 8

  p, admin, /*, * p, anonymous, /article, GET p, anonymous, /article/*, GET p, member, /permission, GET p, member, /article, GET p, member, /article/*, * g, test01, anonymous g, user01, member

  说明：

• line 1: 定义admin角色能访问所有资源

• line 2,3: 定义anonymous角色能查看文章列表、文章详情

• line 4: 定义member角色能获取权限

• line 5,6: 定义member角色能访问文章所有权限

• line 7: 定义test01的角色为anonymous

• line 8: 定义user01的角色为member

定义中间件

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

// enforcer.go

package main

import (
	"github.com/casbin/casbin"
	"github.com/gin-gonic/gin"
	"log"
	"net/http"
)

var enforcer *casbin.Enforcer

// init 初始化
func init()  {
	authEnforcer, err := casbin.NewEnforcerSafe("conf/rbac_model.conf", "conf/policy.csv")
	if err != nil {
		log.Fatal(err)
	}
	enforcer = authEnforcer
}

// GetPermissionsByUser 获取用户权限，用于前端展示
func GetPermissionsByUser(username string) [][]string  {
	return enforcer.GetImplicitPermissionsForUser(username)
}

// CheckPermission 权限校验中间件
func CheckPermission(ctx *gin.Context)  {
    // 获取角色，一般不会通过参数传递角色，而是根据用户当前的token解析得到用户的角色
    // 这里为了演示而简单处理
	role := ctx.Query("role")
	if role == "" {
		role = "anonymous"
	}

    // 根据角色,路由，请求方法校验权限
	hasPermission, err := enforcer.EnforceSafe(role, ctx.Request.URL.Path, ctx.Request.Method)

    // 程序异常
	if err != nil {
		ctx.String(http.StatusInternalServerError, err.Error())
		ctx.Abort()
	}

    // 没有权限
	if !hasPermission {
		ctx.String(http.StatusUnauthorized, "StatusUnauthorized")
		ctx.Abort()
	}

	ctx.Next()
}

web服务

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

package main

import (
	"github.com/gin-gonic/gin"
	"net/http"
)

func main() {
	router := gin.Default()
    // 使用中间件
    router.Use(CheckPermission)
    
    // 根据用户
	router.GET("permission", func(context *gin.Context) {
		context.JSON(http.StatusOK, gin.H{
			"data": GetPermissionsByUser(context.Query("user")),
		})
	})

    // 定义一组文章的接口
	article := router.Group("article")
	{
        // 列表查询
		article.GET("", func(context *gin.Context) {
			context.JSON(http.StatusOK, gin.H{
				"data": "paginate",
			})
		})

        // 查看详情
		article.GET("/:id", func(context *gin.Context) {
			context.JSON(http.StatusOK, gin.H{
				"data": "info:" + context.Param("id"),
			})
		})

        // 更新
		article.PUT("/:id", func(context *gin.Context) {
			context.JSON(http.StatusOK, gin.H{
				"data": "update"+ context.Param("id"),
			})
		})

        // 创建
		article.POST("", func(context *gin.Context) {
			context.JSON(http.StatusOK, gin.H{
				"data": "create",
			})
		})

        // 删除
		article.DELETE("/:id", func(context *gin.Context) {
			context.JSON(http.StatusOK, gin.H{
				"data": "delete"+ context.Param("id"),
			})
		})
	}


	router.Run(":8888")

}

测试

• 启动

  1	go run mian.go

• 测试

  1 2 3 4 5 6 7 8

  curl localhost:8888/article // output: {"data":"paginate"} curl localhost:8888/article/1 // output: {"data":"info:1"} curl localhost:8888/permission // output: StatusUnauthorized curl localhost:8888/permission?role=member&user=user01 // output: {"data":[["member","/permission","GET"],["member","/article","GET"],["member","/article/*","*"]]}

本文介绍golang里也可以像python那样强大的使用爬虫。

什么是爬虫

网络爬虫（又称为网页蜘蛛，网络机器人，在FOAF社区中间，更经常的称为网页追逐者），是一种按照一定的规则，自动地抓取万维网信息的程序或者脚本。另外一些不常使用的名字还有蚂蚁、自动索引、模拟程序或者蠕虫。

说明

爬虫在19年国家颁布《中华人民共和国网络安全法》里，明确指出了非法爬取数据的规定，此文章仅为学习讨论，合法学习并使用爬虫。

爬虫框架

• chromedb
  无外部依赖，直接驱动支持chrome的DevTool协议的浏览器发起http请求

安装：

1

go get -u github.com/chromedb/chromedb

当然你的电脑上还需要安装chrome

示例Demo

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

package main

import (
	"context"
	"github.com/chromedp/chromedp"
	"log"
	"strings"
)

func main() {
	// 如果是windows,出现以下错误时：
	// exec: "google-chrome": executable file not found in %PATH%
	// 需要通过下面方式指定运行环境
	// linux只要安装了chrome,一般不会出现这些问题
	allocCtx, allocCancel := chromedp.NewExecAllocator(context.Background(), chromedp.ExecPath("C:\\Users\\cx\\AppData\\Local\\Google\\Chrome\\Application\\chrome.exe"))
	defer allocCancel()

	// 如果是linux,可以直接使用chrome headless模式:
	// chromedp.NewContext(context.Background())
	ctx, cancel := chromedp.NewContext(allocCtx)
	defer cancel()


	var res string

	// 运行
	err := chromedp.Run(ctx,
		// 指定url
		chromedp.Navigate("https://segmentfault.com/questions"),

		// 通过指定标签页，获取html内容，并将内容写入res变量里
		chromedp.Text(`div[class="copyright"]`, &res, chromedp.NodeVisible, chromedp.BySearch),
	)

	if err != nil {
		log.Fatal(err)
	}

	log.Println(strings.TrimSpace(res))

}

本文介绍golang里使用websocket实现即时通讯。

什么是websocket

WebSocket使得客户端和服务器之间的数据交换变得更加简单，允许服务端主动向客户端推送数据。在WebSocket API中，浏览器和服务器只需要完成一次握手，两者之间就直接可以创建持久性的连接，并进行双向数据传输。

安装包

这是我封装好的包

1

go get github.com/ebar-go/ws

简单示例

基于gin启动websocket服务

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

package main
import (
    "github.com/ebar-go/ws"
    "github.com/gin-gonic/gin"
    "net/http"
)

var m ws.Manager
func init() {
    m = ws.New()
}

func main() {
	// use gin router
	router := gin.Default()
	router.GET("/ws", func(ctx *gin.Context) {
		// get websocket conn
		conn, err := ws.GetUpgradeConnection(ctx.Writer, ctx.Request)
		if err != nil {
			http.NotFound(ctx.Writer, ctx.Request)
			return
		}
        
        // 指定客户端的消息处理方法
		client := ws.NewConnection(conn, func(ctx *ws.Context) string {
            // 返回响应内容
			return ctx.GetMessage()
        })
        
        // listen
        go client.Listen()
        
        // register connection
		m.Register(client)
        
	})
    // start websocket service
	go m.Start()

	_ = router.Run()    
}

测试

使用wscat测试websocket

1
2
3
4
5
6
7
8
9
10
11

// 安装
npm install -g wscat

// 建立连接
wscat -c ws://localhost:8080

// 发送请求
> hello

// 收到的响应
< hello