Fork me on GitHub

服务化04-消息队列kafka

本文介绍在微服务架构中，常用于服务与服务之间的异步通信组件kafka。

kafka

Kafka是由Apache软件基金会开发的一个开源流处理平台，是一种高吞吐量的分布式发布订阅消息系统。

使用场景

  • 在系统或应用程序之间构建可靠的用于传输实时数据的管道，消息队列功能
  • 构建实时的流数据处理程序来变换或处理数据流，数据处理功能

image

通过docker部署kafka

  • 首先需要安装zookeeper

    1
    docker run -d --name zookeeper -p 2181:2181 -v /etc/localtime:/etc/localtime wurstmeister/zookeeper

  • 然后启动kafka容器

    1
    2
    3
    4
    5
    6
    docker run  -d --name kafka -p 9092:9092 \
    -e KAFKA_BROKER_ID=0 \
    -e KAFKA_ZOOKEEPER_CONNECT=172.17.0.1:2181 \
    -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://172.17.0.1:9092 \
     -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 \
     -t wurstmeister/kafka

使用kafka-go

  • 安装

    1
    go get github.com/segmentio/kafka-go

  • 封装客户端

    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

    import (
    	"context"
    	"fmt"
    	"github.com/segmentio/kafka-go"
    	"time"
    )

    type Client struct {
    	host string // kafka地址,一般是:localhost:9092
    	instance *kafka.Conn // 
    }

    // Connect 连接kafka服务
    func (client *Client) Connect( topic string) error {
    	conn, err := kafka.DialLeader(context.Background(), "tcp", client.host, topic, 0)
    	if err != nil {
    		return fmt.Errorf("failed to dial learder: %v", err)
    	}

    	_ = conn.SetWriteDeadline(time.Now().Add(10*time.Second))

    	client.instance = conn
    	return nil
    }

    // NewClient 实例化
    func NewClient(host string) *Client {
    	return &Client{host: host}
    }
more >>

展开全文 >>

Go数据结构与算法(03)-字典树

本文将介绍字典树的相关知识与Golang的实现方式。

Trie

在计算机科学中，trie，又称前缀树或字典树，是一种有序树，用于保存关联数组，其中的键通常是字符串。与二叉查找树不同，键不是直接保存在节点中，而是由节点在树中的位置决定。一个节点的所有子孙都有相同的前缀，也就是这个节点对应的字符串，而根节点对应空字符串。

Trie的典型应用是用于统计，排序和保存大量的字符串（但不仅限于字符串），所以经常被搜索引擎系统用于文本词频统计、敏感词检测、文本提示等。它的优点是：利用字符串的公共前缀来减少查询时间，最大限度地减少无谓的字符串比较，查询效率比哈希树高。

结构

如果插入数据为sex,son,sleep, some,则树结构如下所示：

  • Trie:
    1
    2
    3
    4
    5
    s
    ├── e ── x
    ├── o ── n
    │   └─ m ── e
    └── l ── e ── e ── p
more >>

展开全文 >>

服务化03-Etcd的服务注册与发现

本文介绍如何通过etcd实现服务注册与发现。

Etcd

一个高可用的分布式键值(key-value)数据库。etcd内部采用raft协议作为一致性算法，etcd基于Go语言实现。

一致性算法raft

节点的三种角色：

  • Leader: 负责接收客户端的请求，将日志复制到其他节点。
  • Candidate: 用于选举Leader的一种角色。
  • Follower: 普通节点，负责响应Leader和Candidate节点的请求。

服务注册与发现

service

服务注册

A服务启动时，将当前服务运行的IP和Port提交到Etcd保存，key为ServiceA。

服务发现

B服务需要调用A服务的接口时，去Etcd通过ServiceA这个key，找到存储的服务A的地址，即可发起调用。

服务注销

A服务现在要重启，就要在停止前主动注销。等待重启后再次注册即可。

接下来我们来通过etcd实现：服务注册/服务发现/服务注销

more >>

展开全文 >>

Golang(27)-websocket进阶

之前有简单的介绍过websocket的基本用法，本文将介绍在golang里如何处理大量websocket连接。

初阶

现在有一个即时通讯的需求，我们实现一个简单的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
package main
import (
	"fmt"
	"github.com/gorilla/websocket"
	"log"
	"net/http"
)
func main() {
	http.ListenAndServe(":8085", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		// 升级为websocket连接
		conn, err := NewWebsocketConnection(w, r)
		if err != nil {
			http.NotFound(w, r)
			return
		}
		go func() { // 开启一个协程去处理连接
			defer conn.Close() // 连接结束时需要关闭
			for {
				_, msg, err := conn.ReadMessage() // 接受消息
				if err != nil {
					log.Println("unable to read message:", err.Error())
					return
				}
				// 向客户端回话,或者其他业务逻辑
				reply := fmt.Sprintf("received:%s", string(msg))
				if err := conn.WriteMessage(websocket.TextMessage, []byte(reply)); err != nil {
					log.Printf("unable to send message")
				}
			}
		}()
	}))
}
var u = websocket.Upgrader{CheckOrigin: func(r *http.Request) bool { return true }} // use default options
// WebsocketConn return web socket connection
func NewWebsocketConnection(w http.ResponseWriter, r *http.Request) (*websocket.Conn, error) {
	return u.Upgrade(w, r, nil)

}

通过wscat -c localhost:8085连接到websocket,并发送hello和123,得到测试结果如下：

1
2
3
4
5

> hello
< received:hello
> 123
< received:123

思考

如果我们的服务需要面向100w个用户时，会发生什么情况？

1
2
1.每创建一个websocket连接，按照以上的实现方式，我们就需要创建一个goroutine来接收客户端的信息。一个goroutine大概需要2~8kb的内存
2.如果是同时有100万个连接，假设每个goroutine占用4kb内存，那么内存消耗大概在：4kb*1000000=4G。

光是保持连接，不做任何处理就已经消耗了4G的内存，还是挺恐怖的，所以下面开始介绍用epoll模型来解决这个问题。

more >>

展开全文 >>

Golang26-内存模型

本文介绍Golang的内存模型相关知识。

什么是go的内存模型

指定了一系列条件，在这些条件下，可以保证在协程中对变量的读取操作可以观察到其他协程对同一变量写操作的结果，这就是go内存模型。

为什么需要这些条件？

因为编译器无法保证指令执行顺序与程序书写顺序一致。

如下示例：

1
2
3
4
5
6
7
8
9
10
package main
func main() {
    i := 5
    go func() {
        i = 10
    }()
    for j := 0;j<100;j++ {
        fmt.Println(i)
    }
}

多运行几次，会发现结果可能不同，但可以发现，部分Println打印的i变量依然是5，说明未能立观察到协程对i变量的写操作。

所以我们为了保证协程间的变量读取的可观察，就需要建立Happen Before关系的同步事件。也就是内存模型设定的条件。

Happen Before

以下几种方式，可以建立Happen Before关系的同步事件:

1
2
3
4
5
1.init函数
2.创建/销毁goroutine
3.channel
4.锁
5.Once
more >>

展开全文 >>

kubernetes介绍与安装

本文作为kubernetes系列的开篇，主要介绍一些相关概念。

什么是kubernetes

kubernetes，简称K8s。是一个开源的，用于管理云平台中多个主机上的容器化的应用，Kubernetes的目标是让部署容器化的应用简单并且高效（powerful）,Kubernetes提供了应用部署，规划，更新，维护的一种机制。

在服务化时代，每个公司都会有很多个分散的项目，我们通过k8s来统一管理这些应用，不但可以提高服务的部署效率，还能提高服务的横向扩展能力。

k8s & docker

  • docker: 应用容器引擎，作为目标程序的载体，可移植，达到运行环境一致性。
  • k8s: 容器集群管理系统，实现容器集群的自动化部署，扩缩容，负载均衡等。

在我看来,docker就像“出租车”，k8s就是“出租车管理公司”。

相关概念

Master

集群控制节点。负责整个集群服务的的管理和控制。Mater节点上运行以下组件：

  • kube-apiserver: http服务，对k8s中所有资源进行管理(增删改查)，是集群控制入口进程。kubectl是访问apiserver的客户端工具。

  • kube-control-manager: apiserver为对外开放的应用服务，与之相对的后台服务就是kube-control-manager,负责k8s里所有资源对象的控制。

  • kube-schedule: 资源调度器，负责调度pod到node节点上。

  • etcd: 存储各种资源的状态数据。

Node

除master节点外的其他机器节点。是集群的工作负载节点。Node节点上运行以下组件：

  • kubelet: 负责pod对应的容器创建、启动、停止、资源监控等任务。
  • kube-proxy: 实现service的通信与负载均衡功能。
  • docker: 容器runtime。

Deployment

当我们需要创建pod来运行应用的时候，不是直接创建Pod,而是创建一个Deployment，用它负责创建和更新应用。可以理解为pod的”管家”。

Service

因为pod可能会不停的创建和销毁，其IP也会随之变化，我们想要在外部固定的访问pod上运行的应用，就需要service来暴露pod里的应用。

service定义了一组pod的逻辑集合和访问策略，不管内部的pod如何变化，我们只要保证service的访问方式不变，其内部会通过服务注册/发现来定位到pod应用。

Pod

  • 在k8s里最小的逻辑运行单元，简单的说就是容器集合。
  • 相同pod里运行的多个容器，共享uts、net、ipc空间等。

安装k8s集群环境

通过kind安装k8s cluster,下载：

1
2
3
curl -Lo ./kind https://kind.sigs.k8s.io/dl/v0.10.0/kind-linux-amd64
chmod +x ./kind
mv ./kind /usr/bin/kind

  • 创建集群

    1
    kind create cluster

  • 查看集群信息

    1
    kubectl cluster-info

  • 查看节点

    1
    kubectl get nodes

  • 删除集群

    1
    kind delete cluster

k8s环境已准备就绪。下一章将介绍如何用k8s部署go的web应用，敬请期待！

展开全文 >>

Golang代码优化25-依赖注入

本文介绍在golang中如何通过依赖注入(Dependency Inject，简称DI)管理全局服务。

什么是DI

把有依赖关系的类放到容器中，解析出这些类的实例，就是依赖注入。

DI的作用

  • 反面例子
    现在我们有一个http应用，先来看下常规开发的main.go:
    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
    func main() {
    	// 生成config实例
    	config := NewConfig()
    	// 连接数据库
    	db, err := ConnectDatabase(config)
    	// 判断是否有错误
    	if err != nil {
    		panic(err)
    	}
    	// 生成repository实例，用于获取person数据，参数是db
    	personRepository := repo.NewPersonRepository(db)
    	// 生成service实例，用于调用repository的方法
    	personService := service.NewPersonService(config, personRepository)
    	// 生成http服务实例
    	server := NewServer(config, personService)
    	// 启动http服务
    	server.Run()
    }
    // Server
    type Server struct {
    	config        *config.Config
    	personService *service.PersonService
    }
    // Handler
    func (s *Server) Handler() http.Handler {
    	mux := http.NewServeMux()

    	mux.HandleFunc("/people", s.people)

    	return mux
    }
    // Run
    func (s *Server) Run() {
    	httpServer := &http.Server{
    		Addr:    ":" + s.config.Port,
    		Handler: s.Handler(),
    	}

    	httpServer.ListenAndServe()
    }
    // people
    func (s *Server) people(w http.ResponseWriter, r *http.Request) {
    	people := s.personService.FindAll()
    	bytes, _ := json.Marshal(people)

    	w.Header().Set("Content-Type", "application/json")
    	w.WriteHeader(http.StatusOK)
    	w.Write(bytes)
    }
    // NewServer
    func NewServer(config *config.Config, service *service.PersonService) *Server {
    	return &Server{
    		config:        config,
    		personService: service,
    	}
    }

    // 其他new方法
    func NewConfig() *Config {
        // ...
    }
    func ConnectDatabase(config *config.Config) (*sql.DB, error) {
        // ... 
    }
    func NewPersonRepository(database *sql.DB) *PersonRepository {
        // ... 
    }
    func NewPersonService(config *config.Config, repository *repo.PersonRepository) *PersonService {
        // ...
    }
    直接看main(),你会发现包含清晰的初始化流程。

但是仔细想想，随着业务的扩展，我们如果把所有实例都在main函数里生成，main函数将变得越来越臃肿。

而且这些基础服务的实例，如果在其他包里需要引入，你就得给每个需要用到服务的地方，通过参数的方式传递。类似service.NewPersonService(config, personRepository)方法，将config和personRepository传递到service包。

  • 问题
    • 如何让main函数变得优雅?
    • 如何管理全局服务实例?
    • 如何减少重复实例化对象时传递如config这样的基础实例?
more >>

展开全文 >>

Golang24-mongodb客户端

上一篇文章介绍了如何部署mongodb集群，本文介绍如何用mgo客户端操作mongodb。

安装

1
go get github.com/globalsign/mgo

连接mongo

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import (
	"fmt"
	"github.com/globalsign/mgo"
	"log"
)

func main() {
	dialInfo := &mgo.DialInfo{
		Addrs:    []string{"172.17.0.1:27017","172.17.0.1:27018"},
	}

	session ,err := mgo.DialWithInfo(dialInfo)
	if err != nil {
		log.Fatalf("connect: %v\n", err)
	}
	fmt.Println(session.DatabaseNames())
	log.Println("connect success")
}

查询

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
type User struct {
	Name string `json:"name"`
}

func Run(session *mgo.Session)  {
	collection := session.Copy().DB("demo").C("user")
	// 查看总数
	count, err := collection.Count()
	if err != nil {
		log.Fatalf("count: %v\n", err)
	}
	log.Printf("collection count:%d\n", count)

	// 插入数据
	if err := collection.Insert(&User{Name: "bob"}); err != nil {
		log.Fatalf("insert: %v\n", err)
	}
	log.Println("insert success")

	// 查询数据
	entity := new(User)
	if err := collection.Find(bson.M{"name":"bob"}).One(entity); err != nil {
		log.Fatalf("Find: %v\n", err)
	}
	log.Printf("find user: %s\n", entity.Name)
}

展开全文 >>

服务化02-mongodb集群

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

概念

Replica Set 副本集：一个副本集就是一组 MongoDB 实例组成的集群，由一个主（Primary）服务器和多个备份（Secondary）服务器构成

  • 节点（master）：主节点接收所有写入操作。主节点将对其数据集所做的所有更改记录到其 oplog。
  • 副节点（secondary）：复制主节点的 oplog 并将操作应用到其数据集，如果主节点不可用，一个合格的副节点将被选为新的主节点。
  • 仲裁节点（arbiter）：负载选举，当主节点不可用，它将从副节点中选一个作为主节点。
more >>

展开全文 >>

Golang23-操作Elasticsearch

上一篇文章介绍了Elasticsearch集群相关的知识，本文介绍如何通过golang去处理elasticsearch数据。

安装

1
go get github.com/olivere/elastic/v7

连接es

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import (
	"fmt"
	"github.com/olivere/elastic/v7"
	"log"
)

func main() {
	// 连接
	var host  = "http://127.0.0.1:9200"
	client, err := elastic.NewClient(elastic.SetURL(host), elastic.SetSniff(false))
	if err != nil {
		log.Fatalf("connect elasticsearch: %v", err)
	}
	log.Println("connect success")
	fmt.Println(client)
}

创建索引

1
2
3
4
5
6
7
8
9
10
11
func CreateIndex(client *elastic.Client) {
    ctx := context.Background()
	result, err := client.CreateIndex("blog").Do(ctx)
	if err != nil {
		log.Fatalf("create index: %v", err)
	}
	if !result.Acknowledged {
		log.Fatalf("create index failed")
	}
	log.Println("create index success")
}

插入数据

1
2
3
4
5
6
7
8
func InsertDoc(client *elastic.Client)  {
	indexName := "blog"
	response, err := client.Index().Index(indexName).BodyJson(`{"title":"hello","content":"world","created_at":1616381976}`).Do(ctx)
	if err != nil {
		log.Fatalf("insert document: %v", err)
	}
	fmt.Printf("insert document success,id: %s", response.Id)
}

查询

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
func QueryDoc(client *elastic.Client)  {
	indexName := "blog"
	boolQuery := elastic.NewTermQuery("title", "hello")

	searchResult, err := client.Search(indexName).Query(boolQuery).Do(context.Background())
	if err != nil {
		log.Fatalf("query document: %v", err)
	}
	fmt.Printf("Query took %d milliseconds\n", searchResult.TookInMillis)
	// Number of hits
	if searchResult.Hits != nil {
		fmt.Printf("Found a total of %d tweets\n", searchResult.Hits.TotalHits.Value)

		// Iterate through results
		for _, hit := range searchResult.Hits.Hits {
			// hit.Index contains the name of the index

			// Deserialize hit.Source into a Tweet (could also be just a map[string]interface{}).
			var t map[string]interface{}
			err := json.Unmarshal(hit.Source, &t)
			if err != nil {
				// Deserialization failed
			}

			// Work with tweet
			fmt.Printf("id:%s, title: %s, content: %s\n", hit.Id, t["title"], t["content"])
		}
	} else {
		// No hits
		fmt.Print("Found no tweets\n")
	}
}

更多示例请查看：https://pkg.go.dev/github.com/olivere/elastic

展开全文 >>

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia-plus根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

Golang、云原生、系统架构
现居：鹅厂

联系QQ:774601526

看心情更新博客
谢谢大家