安装版本

Elasticsearch（通过apt方式安装）:

• Elasticsearch 2.3.2

Jar插件:

• elasticsearch-analysis-ik: 1.9.2
• elasticsearch-analysis-pinyin: 1.7.2
• elasticsearch-analysis-stconvert: 1.8.2

Site插件:

• elasticsearch-head: 最新版

各节点服务器

• Node1
  • 集群名：search-1
  • 节点名：node-1
  • 内网IP: 192.168.31.171
• Node2
  • 集群名：search-1
  • 节点名：node-2
  • 内网IP: 192.168.31.221
• Node3
  • 集群名：search-1
  • 节点名：node-3
  • 内网IP: 192.168.31.154

各节点安装的插件

各节点配置

三节点的配置：

• Node 1

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

cluster.name: search-1
node.name: node-1

node.master: true
node.data: true

index.number_of_shards: 3
index.number_of_replicas: 1

network.host: 0.0.0.0                           # 绑定本机所有端口

discovery.zen.ping.multicast.enabled: false     # 禁止多播
discovery.zen.minimum_master_nodes: 2           # 配置最少节点数量，防止脑裂
discovery.zen.ping.unicast.hosts: ['192.168.31.171', '192.168.31.221', '192.168.31.154']

• Node 2

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

cluster.name: search-1
node.name: node-2

node.master: true
node.data: true

index.number_of_shards: 3
index.number_of_replicas: 1

network.host: 0.0.0.0                           # 绑定本机所有端口

discovery.zen.ping.multicast.enabled: false     # 禁止多播
discovery.zen.minimum_master_nodes: 2           # 配置最少节点数量，防止脑裂
discovery.zen.ping.unicast.hosts: ['192.168.31.171', '192.168.31.221', '192.168.31.154']

• Node 3

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

cluster.name: search-1
node.name: node-3

node.master: true
node.data: true

index.number_of_shards: 3
index.number_of_replicas: 1

network.host: 0.0.0.0                           # 绑定本机所有端口

discovery.zen.ping.multicast.enabled: false     # 禁止多播
discovery.zen.minimum_master_nodes: 2           # 配置最少节点数量，防止脑裂
discovery.zen.ping.unicast.hosts: ['192.168.31.171', '192.168.31.221', '192.168.31.154']

常用操作

上传插件

1
2
3
4
5
6
7
8
9
10
11

$ scp -r -i ~/dev.pem elasticsearch-analysis-ik-1.9.2 ubuntu@192.168.31.171:~/
$ scp -r -i ~/dev.pem elasticsearch-analysis-pinyin-1.7.2 ubuntu@192.168.31.171:~/
$ scp -r -i ~/dev.pem elasticsearch-analysis-stconvert-1.8.2 ubuntu@192.168.31.171:~/

$ scp -r -i ~/dev.pem elasticsearch-analysis-ik-1.9.2 ubuntu@192.168.31.221:~/
$ scp -r -i ~/dev.pem elasticsearch-analysis-pinyin-1.7.2 ubuntu@192.168.31.221:~/
$ scp -r -i ~/dev.pem elasticsearch-analysis-stconvert-1.8.2 ubuntu@192.168.31.221:~/

$ scp -r -i ~/dev.pem elasticsearch-analysis-ik-1.9.2 ubuntu@192.168.31.154:~/
$ scp -r -i ~/dev.pem elasticsearch-analysis-pinyin-1.7.2 ubuntu@192.168.31.154:~/
$ scp -r -i ~/dev.pem elasticsearch-analysis-stconvert-1.8.2 ubuntu@192.168.31.154:~/

拷贝插件到/usr/share/elasticsearch/plugins目录

1
2
3
4
5
6
7

$ mv elasticsearch-analysis-ik-1.9.2 /usr/share/elasticsearch/plugins/
$ mv elasticsearch-analysis-pinyin-1.7.2 /usr/share/elasticsearch/plugins/
$ mv elasticsearch-analysis-stconvert-1.8.2 /usr/share/elasticsearch/plugins/

$ chown -R elasticsearch:elasticsearch elasticsearch-analysis-ik-1.9.2/             # 修改插件访问权限，允许elasticsearch服务访问插件
$ chown -R elasticsearch:elasticsearch elasticsearch-analysis-pinyin-1.7.2/         # 修改插件访问权限，允许elasticsearch服务访问插件
$ chown -R elasticsearch:elasticsearch elasticsearch-analysis-stconvert-1.8.2/      # 修改插件访问权限，允许elasticsearch服务访问插件

参考ElasticSearch基础概念:

Near Realtime:

there is a slight latency (normally one second) from the time you index a document until the time it becomes searchable.

Cluster

A cluster is a collection of one or more nodes (servers) that together holds your entire data and provides federated indexing and search capabilities across all nodes. A cluster is identified by a unique name which by default is “elasticsearch”. This name is important because a node can only be part of a cluster if the node is set up to join the cluster by its name.

Make sure that you don’t reuse the same cluster names in different environments, otherwise you might end up with nodes joining the wrong cluster. For instance you could use logging-dev, logging-stage, and logging-prod for the development, staging, and production clusters.

Note that it is valid and perfectly fine to have a cluster with only a single node in it. Furthermore, you may also have multiple independent clusters each with its own unique cluster name.

Node

A node is a single server that is part of your cluster, stores your data, and participates in the cluster’s indexing and search capabilities. Just like a cluster, a node is identified by a name which by default is a random Marvel character name that is assigned to the node at startup. You can define any node name you want if you do not want the default. This name is important for administration purposes where you want to identify which servers in your network correspond to which nodes in your Elasticsearch cluster.

A node can be configured to join a specific cluster by the cluster name. By default, each node is set up to join a cluster named elasticsearch which means that if you start up a number of nodes on your network and—assuming they can discover each other—they will all automatically form and join a single cluster named elasticsearch.

In a single cluster, you can have as many nodes as you want. Furthermore, if there are no other Elasticsearch nodes currently running on your network, starting a single node will by default form a new single-node cluster named elasticsearch.

Index

An index is a collection of documents that have somewhat similar characteristics. For example, you can have an index for customer data, another index for a product catalog, and yet another index for order data. An index is identified by a name (that must be all lowercase) and this name is used to refer to the index when performing indexing, search, update, and delete operations against the documents in it.

In a single cluster, you can define as many indexes as you want.

Type

Within an index, you can define one or more types. A type is a logical category/partition of your index whose semantics is completely up to you. In general, a type is defined for documents that have a set of common fields. For example, let’s assume you run a blogging platform and store all your data in a single index. In this index, you may define a type for user data, another type for blog data, and yet another type for comments data.

Document

A document is a basic unit of information that can be indexed. For example, you can have a document for a single customer, another document for a single product, and yet another for a single order. This document is expressed in JSON (JavaScript Object Notation) which is an ubiquitous internet data interchange format.

Within an index/type, you can store as many documents as you want. Note that although a document physically resides in an index, a document actually must be indexed/assigned to a type inside an index.

Shards & Replicas

An index can potentially store a large amount of data that can exceed the hardware limits of a single node. For example, a single index of a billion documents taking up 1TB of disk space may not fit on the disk of a single node or may be too slow to serve search requests from a single node alone.

To solve this problem, Elasticsearch provides the ability to subdivide your index into multiple pieces called shards. When you create an index, you can simply define the number of shards that you want. Each shard is in itself a fully-functional and independent “index” that can be hosted on any node in the cluster.

Sharding is important for two primary reasons:

• It allows you to horizontally split/scale your content volume
• It allows you to distribute and parallelize operations across shards (potentially on multiple nodes) thus increasing performance/throughput
  The mechanics of how a shard is distributed and also how its documents are aggregated back into search requests are completely managed by Elasticsearch and is transparent to you as the user.

In a network/cloud environment where failures can be expected anytime, it is very useful and highly recommended to have a failover mechanism in case a shard/node somehow goes offline or disappears for whatever reason. To this end, Elasticsearch allows you to make one or more copies of your index’s shards into what are called replica shards, or replicas for short.

Replication is important for two primary reasons:

• It provides high availability in case a shard/node fails. For this reason, it is important to note that a replica shard is never allocated on the same node as the original/primary shard that it was copied from.
• It allows you to scale out your search volume/throughput since searches can be executed on all replicas in parallel.

To summarize, each index can be split into multiple shards. An index can also be replicated zero (meaning no replicas) or more times. Once replicated, each index will have primary shards (the original shards that were replicated from) and replica shards (the copies of the primary shards). The number of shards and replicas can be defined per index at the time the index is created. After the index is created, you may change the number of replicas dynamically anytime but you cannot change the number shards after-the-fact.

By default, each index in Elasticsearch is allocated 5 primary shards and 1 replica which means that if you have at least two nodes in your cluster, your index will have 5 primary shards and another 5 replica shards (1 complete replica) for a total of 10 shards per index.

Elasticsearch的安装有两种方式

• 通过tar.gz压缩包方式安装，参考ElasticSearch Reference 2.3 - Installation
• 通过apt命令安装，参考ElasticSearch Reference 2.3 - Repositories

通过tar.gz压缩包安装步骤

• SSH远程连接Ubuntu Server，ssh ethancai@###.###.###.###
• 下载Elasticsearch
  • mkdir ~/Downloads && cd ~/Downloads && curl -L -O https://download.elastic.co/elasticsearch/release/org/elasticsearch/distribution/tar/elasticsearch/2.3.1/elasticsearch-2.3.1.tar.gz
• 解压
  • tar -xzvf elasticsearch-2.3.1.tar.gz && sudo mv elasticsearch-2.3.1 /var/
• 运行
  • cd /var/elasticsearch-2.3.1/bin
  • ./elasticsearch，或者以指定cluster和node名称的方式运行./elasticsearch --cluster.name my_cluster_name --node.name my_node_name

通过apt命令安装

• Download and install the Public Signing Key
  • wget -qO - https://packages.elastic.co/GPG-KEY-elasticsearch | sudo apt-key add -
• Save the repository definition
  • echo "deb http://packages.elastic.co/elasticsearch/2.x/debian stable main" | sudo tee -a /etc/apt/sources.list.d/elasticsearch-2.x.list
• Install elasticsearch, 注意这里需要指定安装版本
  • sudo apt-get update && sudo apt-get install elasticsearch=2.3.2
• Configure Elasticsearch to automatically start during bootup
  • sudo update-rc.d elasticsearch defaults 95 10

验证安装是否成功

By default, Elasticsearch uses port 9200 to provide access to its REST API. 你可以通过下面命令验证是否启动成功：

• curl 'localhost:9200/_cat/health?v'
• curl 'localhost:9200/_cat/nodes?v'
• 验证后记得重启，sudo shutdown -r 0

默认目录

不同安装方式，默认的目录位置不一样，参考ElasticSearch Reference 2.3 - Directory Layout

通过tar.gz方式安装的默认目录如下:

通过apt方式安装的默认目录如下：

安装Elasticsearch插件

Plugins are a way to enhance the core Elasticsearch functionality in a custom manner. They range from adding custom mapping types, custom analyzers, native scripts, custom discovery and more.

There are three types of plugins:

• Java Plugins: These plugins contain only JAR files, and must be installed on every node in the cluster. After installation, each node must be restarted before the plugin becomes visible.
• Site Plugins: These plugins contain static web content like Javascript, HTML, and CSS files, that can be served directly from Elasticsearch. Site plugins may only need to be installed on one node, and do not require a restart to become visible.
  • The content of site plugins is accessible via a URL like: http://yournode:9200/_plugin/[plugin name]
• Mixed Plugins: Mixed plugins contain both JAR files and web content.

安装Site Plugins

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

$ sudo su -
$ cd /

# 查看已安装插件
$ /usr/share/elasticsearch/bin/plugin list

# 安装elasticsearch-head，访问http://localhost:9200/_plugin/head
$ /usr/share/elasticsearch/bin/plugin install mobz/elasticsearch-head

# 安装elasticsearch-kopf，访问http://localhost:9200/_plugin/kopf
$ /usr/share/elasticsearch/bin/plugin install lmenezes/elasticsearch-kopf

# 安装elastichq，访问http://localhost:9200/_plugin/hq
$ /usr/share/elasticsearch/bin/plugin install royrusso/elasticsearch-HQ

# 安装elasticsearch-inquisitor，访问http://localhost:9200/_plugin/elasticsearch-inquisitor
$ /usr/share/elasticsearch/bin/plugin install polyfractal/elasticsearch-inquisitor

安装Jar Plugins

安装elasticsearch-analysis-ik

安装前请注意和Elasticsearch的适配版本，安装elasticsearch-analysis-ik有两种方式：

• 通过源代码手动编译安装
  • 请先安装apache maven
  • 下载Apache Maven的源代码
    • git clone https://github.com/medcl/elasticsearch-analysis-ik.git
  • 编译打包
    • cd [elasticsearch-analysis-ik所在目录]
    • git checkout v1.9.2
    • mvn package
  • 解压编译生成的打包文件target/releases/elasticsearch-analysis-ik-{version}.zip，并拷贝解压后的文件到Elasticsearch的plugins目录
    • unzip elasticsearch-analysis-ik-{version}.zip -d elasticsearch-analysis-ik-{version}
    • scp -r elasticsearch-analysis-ik-{version} ubuntu@xxx.xxx.xxx.xxx:/var/tmp/
    • cp -R /var/tmp/elasticsearch-analysis-ik-{version} /usr/share/elasticsearch/plugins
    • chown -R elasticsearch:elasticsearch /usr/share/elasticsearch/plugins/elasticsearch-analysis-ik-{version}
  • 重启elasticsearch服务
    • service elasticsearch restart
• 直接下载安装程序包
  • 下载指定版本的打包文件
  • 后续步骤同上

注意：

• v2.2.1版本以后安装elasticsearch-analysis-ik并不需要修改elasticsearch.yml配置文件

安装elasticsearch-analysis-pinyin

安装前请注意和Elasticsearch的适配版本, 安装elasticsearch-analysis-pinyin有两种方式：

• 通过源代码手动编译安装
  • 具体操作参考elasticsearch-analysis-ik的安装步骤
• 直接下载安装程序包

安装elasticsearch-analysis-stconvert

安装前请注意和Elasticsearch的适配版本, 安装elasticsearch-analysis-stconvert有两种方式：

• 通过源代码手动编译安装
  • 具体操作参考elasticsearch-analysis-ik的安装步骤
• 直接下载安装程序包

总结

建议Oracle JDK和Elasticsearch均使用apt安装方式安装，原因如下：

• 安装方便
• 默认目录的配置更规范
• 和Elasticsearch官方的Docker镜像的目录结构一致，未来切换到Docker方式部署障碍少，参考Dockerfile

如何选择JDK版本

Elasticsearch 2.3官方文档中推荐的JDK版本是“Oracle JDK version 1.8.0_73”，最低Java 7.

Oracle的JDK的安装方式

Oracle JDK的安装方式有两种方式

• 手动安装，参考How to Install Oracle Java JDK on Ubuntu Linux
• 通过apt命令安装，参考Installing the oracle JDK

手动安装步骤

• 从这里下载JDK 8u91
• 拷贝下载安装包到~/Download/目录下：’scp ~/Downloads/jdk-8u91-linux-x64.tar.gz ethancai@172.16.210.149:~/Downloads/‘
• SSH远程连接Ubuntu Server，ssh ethancai@172.16.210.149
• 删除OpenJDK
  • sudo apt-get purge openjdk-\*
• 拷贝安装文件到安装目录
  • sudo mkdir -p /usr/local/java
  • sudo cp -r ~/Downloads/jdk-8u91-linux-x64.tar.gz /usr/local/java/
• 解压
  • cd /usr/local/java
  • sudo tar -xvzf jdk-8u91-linux-x64.tar.gz

• 修改登录式Shell的全局启动配置

  • sudo vim /etc/profile，然后将如下内容拷贝到文件中

    1 2 3 4 5 6

    JAVA_HOME=/usr/local/java/jdk1.8.0_91 JRE_HOME=$JAVA_HOME/jre PATH=$PATH:$JAVA_HOME/bin:$JRE_HOME/bin export JAVA_HOME export JRE_HOME export PATH

• Notify the system that JRE/JDK/Java Web Start is available for use

  • sudo update-alternatives --install "/usr/bin/java" "java" "/usr/local/java/jdk1.8.0_91/bin/java" 1
  • sudo update-alternatives --install "/usr/bin/javac" "javac" "/usr/local/java/jdk1.8.0_91/bin/javac" 1
  • sudo update-alternatives --install "/usr/bin/javaws" "javaws" "/usr/local/java/jdk1.8.0_91/bin/javaws" 1
• Reload your system wide PATH
  • source /etc/profile

通过apt命令安装步骤

• 此命令目前能安装的jdk8最新版本是jdk-8u77-linux-x64
• sudo apt-get purge openjdk-\*
• sudo add-apt-repository ppa:webupd8team/java
• sudo apt-get update
• sudo apt-get install oracle-java8-installer

验证安装是否成功

• java -version
• javac -version
• 重启sudo shutdown -r 0

前言

一个人不会两次掉进同一个坑里，但是如果他（她）忘记了坑的位置，那就不一定了。

这篇文章记录了最近使用Golang处理JSON遇到的一些坑。

坑

1号坑：omitempty的行为

C#中最常用的JSON序列化类库Newtonsoft.Json中，把一个类的实例序列化成JSON，如果我们不想让某个属性输出到JSON中，可以通过property annotation或者ShouldSerialize method等方法，告知序列化程序。如下：

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

// 通过ShouldSerialize method指示不要序列化ObsoleteSetting属性
class Config
{
    public Fizz ObsoleteSetting { get; set; }

    public bool ShouldSerializeObsoleteSetting()
    {
        return false;
    }
}

// 通过JsonIgnore的annotation指示不需要序列化ObsoleteSetting属性
class Config
{
    [JsonIgnore]
    public Fizz ObsoleteSetting { get; set; }

    public Bang ReplacementSetting { get; set; }
}

关于Newtonsoft.Json的Conditional Property Serialization的更多内容参考：

• Conditional Property Serialization
• Making a property deserialize but not serialize with json.net

开始使用Golang的时候，以为omitempty的行为和C#中一样用来控制是否序列化字段，结果使用的时候碰了一头钉子。回头阅读encoding/json package的官方文档，找到对omitempty行为的描述：

Struct values encode as JSON objects. Each exported struct field becomes a member of the object unless

• the field’s tag is “-“, or
• the field is empty and its tag specifies the “omitempty” option.

The empty values are false, 0, any nil pointer or interface value, and any array, slice, map, or string of length zero. The object’s default key string is the struct field name but can be specified in the struct field’s tag value. The “json” key in the struct field’s tag value is the key name, followed by an optional comma and options. Examples:

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

> // Field is ignored by this package. > Field int `json:"-"` > > // Field appears in JSON as key "myName". > Field int `json:"myName"` > > // Field appears in JSON as key "myName" and > // the field is omitted from the object if its value is empty, > // as defined above. > Field int `json:"myName,omitempty"` > > // Field appears in JSON as key "Field" (the default), but > // the field is skipped if empty. > // Note the leading comma. > Field int `json:",omitempty"` >

Golang中，如果指定一个field序列化成JSON的变量名字为-，则序列化的时候自动忽略这个field。这种用法，才是和上面JsonIgnore的用法的作用是一样的。

而omitempty的作用是当一个field的值是empty的时候，序列化JSON时候忽略这个field（Newtonsoft.Json的类似用法参考这里和例子）。这里需要注意的是关于emtpty的定义：

The empty values are false, 0, any nil pointer or interface value, and any array, slice, map, or string of length zero.

通过下面的例子，来加深对empty values的了解：

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

package main

import (
    "bytes"
    "encoding/json"
    "log"
    "os"
)

type S1 struct {
    I1 int
    I2 int `json:",omitempty"`

    F1 float64
    F2 float64 `json:",omitempty"`

    S1 string
    S2 string `json:",omitempty"`

    B1 bool
    B2 bool `json:",omitempty"`

    Slice1 []int
    Slice2 []int `json:",omitempty"`
    Slice3 []int `json:",omitempty"`

    Map1 map[string]string
    Map2 map[string]string `json:",omitempty"`
    Map3 map[string]string `json:",omitempty"`

    O1 interface{}
    O2 interface{} `json:",omitempty"`
    O3 interface{} `json:",omitempty"`
    O4 interface{} `json:",omitempty"`
    O5 interface{} `json:",omitempty"`
    O6 interface{} `json:",omitempty"`
    O7 interface{} `json:",omitempty"`
    O8 interface{} `json:",omitempty"`

    P1 *int
    P2 *int               `json:",omitempty"`
    P3 *int               `json:",omitempty"`
    P4 *float64           `json:",omitempty"`
    P5 *string            `json:",omitempty"`
    P6 *bool              `json:",omitempty"`
    P7 *[]int             `json:",omitempty"`
    P8 *map[string]string `json:",omitempty"`
}

func main() {

    p3 := 0
    p4 := float64(0)
    p5 := ""
    p6 := false
    p7 := []int{}
    p8 := map[string]string{}

    s1 := S1{
        I1: 0,
        I2: 0,

        F1: 0,
        F2: 0,

        S1: "",
        S2: "",

        B1: false,
        B2: false,

        Slice1: []int{},
        Slice2: nil,
        Slice3: []int{},

        Map1: map[string]string{},
        Map2: nil,
        Map3: map[string]string{},

        O1: nil,
        O2: nil,
        O3: int(0),
        O4: float64(0),
        O5: "",
        O6: false,
        O7: []int{},
        O8: map[string]string{},

        P1: nil,
        P2: nil,
        P3: &p3,
        P4: &p4,
        P5: &p5,
        P6: &p6,
        P7: &p7,
        P8: &p8,
    }

    b, err := json.Marshal(s1)
    if err != nil {
        log.Printf("marshal error: %v", err)
        return
    }

    var out bytes.Buffer
    json.Indent(&out, b, "", "\t")
    out.WriteTo(os.Stdout)
    //Output:
    //{
    //	"I1": 0,
    //	"F1": 0,
    //	"S1": "",
    //	"B1": false,
    //	"Slice1": [],
    //	"Map1": {},
    //	"O1": null,
    //	"O3": 0,
    //	"O4": 0,
    //	"O5": "",
    //	"O6": false,
    //	"O7": [],
    //	"O8": {},
    //	"P1": null,
    //	"P2": 0
    //}%
}

点击这里执行上面的程序

关于empty value的定义，这里面隐藏了一些坑。下面通过一个例子来说明。

假设我们有一个社交类App，通过Restful API形式从服务端获取当前登录用户基本信息及粉丝数量。如果服务端对Response中User对象的定义如下：

1
2
3
4
5

type User struct {
    ID        int `json:"id"`                  // 用户id
    // 其它field
    FansCount int `json:"fansCount,omitempty"` // 粉丝数
}

如果正在使用App时一个还没有粉丝的用户，访问Restful API的得到Response如下：

1
2
3
4

{
    "id": 1000386,
    ...
}

这时候你会发现Response的User对象中没有fansCount，因为fansCount是个int类型且值为0，序列化的时候会被忽略。语义上，User对象中没有fansCount应该理解为粉丝数量未知，而不是没有粉丝。

如果我们希望做到能够区分粉丝数未知和没有粉丝两种情况，需要修改User的定义：

1
2
3
4
5

type User struct {
    ID        int  `json:"id"`                  // 用户id
    // 其它field
    FansCount *int `json:"fansCount,omitempty"` // 粉丝数
}

将FansCount修改为指针类型，如果为nil，表示粉丝数未知；如果为整数（包括0），表示粉丝数。

这么修改语义上没有漏洞了，但是代码中要给FansCount赋值的时候却要多一句废话。必须先将从数据源查询出粉丝数赋给一个变量，然后再将变量的指针传给FansCount。代码读起来实在是啰嗦：

1
2
3
4
5
6
7
8

// FansCount是int类型时候
user := dataAccess.GetUserInfo(userId)
user.FansCount = dataAccess.GetFansCount(userId)

// FansCount是*int类型的时候
user := dataAccess.GetUserInfo(userId)
fansCount := dataAccess.GetFansCount(userId)
user.FansCount = &fansCount

2号坑：JSON反序列化成interface{}对Number的处理

JSON的规范中，对于数字类型，并不区分是整型还是浮点型。

对于如下JSON文本:

1
2
3
4

{
    "name": "ethancai",
    "fansCount": 9223372036854775807
}

如果反序列化的时候指定明确的结构体和变量类型

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

package main

import (
    "encoding/json"
    "fmt"
)

type User struct {
    Name      string
    FansCount int64
}

func main() {
    const jsonStream = `
        {"name":"ethancai", "fansCount": 9223372036854775807}
    `
    var user User  // 类型为User
    err := json.Unmarshal([]byte(jsonStream), &user)
    if err != nil {
        fmt.Println("error:", err)
    }

    fmt.Printf("%+v \n", user)
}
// Output:
//  {Name:ethancai FansCount:9223372036854775807}

点击这里执行上面的程序

如果反序列化不指定结构体类型或者变量类型，则JSON中的数字类型，默认被反序列化成float64类型：

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

package main

import (
    "encoding/json"
    "fmt"
    "reflect"
)

func main() {
    const jsonStream = `
        {"name":"ethancai", "fansCount": 9223372036854775807}
    `
    var user interface{}  // 不指定反序列化的类型
    err := json.Unmarshal([]byte(jsonStream), &user)
    if err != nil {
        fmt.Println("error:", err)
    }
    m := user.(map[string]interface{})

    fansCount := m["fansCount"]

    fmt.Printf("%+v \n", reflect.TypeOf(fansCount).Name())
    fmt.Printf("%+v \n", fansCount.(float64))
}

// Output:
// 	float64
//  	9.223372036854776e+18

点击这里执行上面的程序

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

package main

import (
    "encoding/json"
    "fmt"
)

type User struct {
    Name      string
    FansCount interface{}  // 不指定FansCount变量的类型
}

func main() {
    const jsonStream = `
        {"name":"ethancai", "fansCount": 9223372036854775807}
    `
    var user User
    err := json.Unmarshal([]byte(jsonStream), &user)
    if err != nil {
        fmt.Println("error:", err)
    }

    fmt.Printf("%+v \n", user)
}

// Output:
// 	{Name:ethancai FansCount:9.223372036854776e+18}

点击这里执行上面的程序

从上面的程序可以发现，如果fansCount精度比较高，反序列化成float64类型的数值时存在丢失精度的问题。

如何解决这个问题，先看下面程序：

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 (
    "encoding/json"
    "fmt"
    "reflect"
    "strings"
)

func main() {
    const jsonStream = `
        {"name":"ethancai", "fansCount": 9223372036854775807}
    `

    decoder := json.NewDecoder(strings.NewReader(jsonStream))
    decoder.UseNumber()    // UseNumber causes the Decoder to unmarshal a number into an interface{} as a Number instead of as a float64.

    var user interface{}
    if err := decoder.Decode(&user); err != nil {
        fmt.Println("error:", err)
            return
        }

    m := user.(map[string]interface{})
    fansCount := m["fansCount"]
    fmt.Printf("%+v \n", reflect.TypeOf(fansCount).PkgPath() + "." + reflect.TypeOf(fansCount).Name())

     v, err := fansCount.(json.Number).Int64()
    if err != nil {
        fmt.Println("error:", err)
            return
    }
    fmt.Printf("%+v \n", v)
}

// Output:
// 	encoding/json.Number
// 	9223372036854775807

点击这里执行上面的程序

上面的程序，使用了func (*Decoder) UseNumber方法告诉反序列化JSON的数字类型的时候，不要直接转换成float64，而是转换成json.Number类型。json.Number内部实现机制是什么，我们来看看源码：

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

// A Number represents a JSON number literal.
type Number string

// String returns the literal text of the number.
func (n Number) String() string { return string(n) }

// Float64 returns the number as a float64.
func (n Number) Float64() (float64, error) {
    return strconv.ParseFloat(string(n), 64)
}

// Int64 returns the number as an int64.
func (n Number) Int64() (int64, error) {
    return strconv.ParseInt(string(n), 10, 64)
}

json.Number本质是字符串，反序列化的时候将JSON的数值先转成json.Number，其实是一种延迟处理的手段，待后续逻辑需要时候，再把json.Number转成float64或者int64。

对比其它语言，Golang对JSON反序列化处理真是易用性太差（“蛋疼”）。

JavaScript中所有的数值都是双精度浮点数（参考这里），反序列化JSON的时候不用考虑数值类型匹配问题。这里多说两句，JSON的全名JavaScript Object Notation（从名字上就能看出和JavaScript的关系非常紧密），发明人是Douglas Crockford，如果你自称熟悉JavaScript而不知道Douglas Crockford是谁，就像是自称是苹果粉丝却不知道乔布斯是谁。

C#语言的第三方JSON处理library Json.NET反序列化JSON对数值的处理也比Golang要优雅的多：

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

using System;
using Newtonsoft.Json;

public class Program
{
    public static void Main()
    {
        string json = @"{
  'Name': 'Ethan',
  'FansCount': 121211,
  'Price': 99.99
}";

        Product m = JsonConvert.DeserializeObject<Product>(json);

        Console.WriteLine(m.FansCount);
        Console.WriteLine(m.FansCount.GetType().FullName);

        Console.WriteLine(m.Price);
        Console.WriteLine(m.Price.GetType().FullName);

    }
}

public class Product
{
    public string Name
    {
        get;
        set;
    }

    public object FansCount
    {
        get;
        set;
    }

    public object Price
    {
        get;
        set;
    }
}

// Output:
//      121211
//      System.Int64
//      99.99
//      System.Double

点击这里执行上面的程序

Json.NET在反序列化的时候自动识别数值是浮点型还是整型，这一点对开发者非常友好。

3号坑：选择什么格式表示日期

JSON的规范中并没有日期类型，不同语言的library对日期序列化的处理也不完全一致：

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

package main

import (
    "encoding/json"
    "fmt"
    "os"
    "time"
)

func main() {
    type Product struct {
        Name      string
        CreatedAt time.Time
    }
    pdt := Product{
        Name:      "Reds",
        CreatedAt: time.Now(),
    }
    b, err := json.Marshal(pdt)
    if err != nil {
        fmt.Println("error:", err)
    }
    os.Stdout.Write(b)
}
// Output
//      {"Name":"Reds","CreatedAt":"2016-06-27T07:40:54.69292134+08:00"}

JavaScript语言：

1
2
3
4
5

➜  ~ node
> var jo = { name: "ethan", createdAt: Date.now() };
undefined
> JSON.stringify(jo)
'{"name":"ethan","createdAt":1466984665633}'

C#语言：

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

using System;
using Newtonsoft.Json;

public class Program
{
    public static void Main()
    {
        Product product = new Product();
        product.Name = "Apple";
        product.CreatedAt = DateTime.Now;

        string json = JsonConvert.SerializeObject(product,
                            Newtonsoft.Json.Formatting.Indented,
                            new JsonSerializerSettings {
                                NullValueHandling = NullValueHandling.Ignore
                            });
        Console.WriteLine(json);
    }
}

public class Product
{
    public string Name
    {
        get;
        set;
    }

    public DateTime CreatedAt
    {
        get;
        set;
    }
}
// Output:
//      {
//        "Name": "Apple",
//        "CreatedAt": "2016-06-26T23:46:57.3244307+00:00"
//      }

Go的encoding/json package、C#的Json.NET默认把日期类型序列化成ISO 8601标准的格式，JavaScript默认把Date序列化从1970年1月1日0点0分0秒的毫秒数。但JavaScript的dateObj.toISOString()能够将日期类型转成ISO格式的字符串，Date.parse(dateString)方法能够将ISO格式的日期字符串转成日期。

个人认为ISO格式的日期字符串可读性更好，但序列化和反序列化时的性能应该比整数更低。这一点从Go语言中time.Time的定义看出来。

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

type Time struct {
    // sec gives the number of seconds elapsed since
    // January 1, year 1 00:00:00 UTC.
    sec int64

    // nsec specifies a non-negative nanosecond
    // offset within the second named by Seconds.
    // It must be in the range [0, 999999999].
    nsec int32

    // loc specifies the Location that should be used to
    // determine the minute, hour, month, day, and year
    // that correspond to this Time.
    // Only the zero Time has a nil Location.
    // In that case it is interpreted to mean UTC.
    loc *Location
}

具体选择哪种形式在JSON中表示日期，有如下几点需要注意：

• 选择标准格式。曾记得.NET Framework官方序列化JSON的方法中，会把日期转成如"\/Date(1343660352227+0530)\/"的专有格式，这样的专有格式对跨语言的访问特别不友好。
• 如果你倾向性能，可以使用整数。如果你倾向可读性，可以使用ISO字符串。
• 如果使用整数表示日期，而你的应用又是需要支持跨时区的，注意一定要是从1970-1-1 00:00:00 UTC开始计算的毫秒数，而不是当前时区的1970-1-1 00:00:00。

参考

文章：

• package encoding/json in Go
• http://docs.studygolang.com/src/encoding/json/example_test.go
• The Go Blog: JSON and Go
• Go by example: JSON
• JSON decoding in Go
• go and json
• Decode JSON Documents In Go
• ffjson: faster JSON serialization for Golang
• Serialization in Go

第三方类库：

• ffjson: faster JSON serialization for Go
• go-simplejson: a Go package to interact with arbitrary JSON
• Jason: Easy-to-use JSON Library for Go
• easyjson
• gabs
• jsonparser

工具：

• JSON-to-Go: instantly converts JSON into a Go type definition

目录

• 前言
• dlog的用途
• 对dlog的一些非功能性需求
• 碰到问题及解决方案
  • 何时使用panic，何时使用return error
  • 如何实现一个logger只能接收对应类型的data log
  • 如何实现批量发送data log
  • 如何实现对Logger.Log方法的调用超时机制
  • 如何在logger没有收到新msg情况下，保证buf中的数据依然会定期发送给AWS Kinesis
  • 如何向程序外部暴露运行指标
  • 如何在单元测试中实现Setup和TearDown
  • 如何实现kinesisMock
  • 如何模拟AWS Kinesis响应慢或者不可用
  • 提交到代码库中的测试代码是否可以保留log.Print
• 踩过的一些坑
• 未来可以优化的地方
• 参考

前言

本文记录了前段时间我和王益使用Go语言合作开发一个log组件dlog的过程中学到的一些知识。在整个合作开发的过程中，王益严谨认真的态度，对开发质量的严格要求，给我留下了极其深刻的印象。能够和王益这样的顶级工程师切磋技艺，对我学习Go语言帮助非常大。也谨以此文表达对王益的感谢。

注：本文假设读者已经对Go语法已经有基本了解。

dlog的用途

首先引用项目readme文档的第一段文字介绍一下dlog的用途：

dlog is a Go package for distributed structure logging using Amazon AWS Kinesis/Firehose.

更多介绍和设计请阅读readme文档

dlog主要是用来记录程序的data log的这样一个Golang package，那什么是data log？这里先简要解释一下。一般程序运行过程中主要产生两类日志：

• status log：主要用于帮助调试、定位程序Bug、或者找到性能瓶颈，比如方法调用日志、错误日志、方法执行时间日志等
• data log：主要用于记录用户行为，收集的data log用于后期的个性化搜索、智能推荐等，比如搜索行为、点击行为等

对dlog的一些非功能性需求

• 每一种类型的data log对应一种logger，一个logger只能记录对应类型的data log
• dlog内部发生的错误，不能影响调用的程序代码的执行
  • 应考虑到AWS Kinesis服务响应慢或者不可用的场景（暂未实现）
• 程序代码中通过调用dlog的方法记录data log，dlog的方法不能阻塞调用的程序代码的执行（这一点dlog暂时未满足要求，需要后期改进）
• AWS Kinesis提供两个API接收数据，一个是PutRecord, 另一个是PutRecords，为了减少对Kinesis的调用次数，采用后者批量发送data log
  • PutRecords对一次调用的record数量限制是500，每个record大小必须小于等于1MB，整个request的大小必须小于等于5MB
  • 每一个Kinesis Stream能够承受的最大TPS和写数据量，与这个stream拥有的shard的数量有关。一个shard支持最大TPS是1000 records per second， 写数据量是1MB per second
• 通过单元测试保证功能正确性

碰到问题及解决方案

何时使用panic，何时使用return error

先看看panic和return error的执行机制。

panic的执行机制

panic会中断当前goroutine的执行，如果不对panic的错误进行recover，那么整个进程都会崩溃。

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

package main

import (
    "fmt"
    "log"
    "time"
)

func main() {
    go func() {
        log.Panic("some error before work2")
        fmt.Println("do some work2")
    }()

    time.Sleep(time.Second)
    fmt.Println("do some work1")
}

执行上面代码请点击这里

可以通过recover捕捉当前goroutine中panic的错误并进行错误处理，整个进程的正常运行不受影响。

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

package main

import (
    "fmt"
    "log"
    "time"
)

func main() {
    go func() {
        defer func() {
            if err := recover(); err != nil {
                log.Printf("error: %v", err)
            }
        }()
        log.Panic("some error before work2")
        fmt.Println("do some work2")
    }()

    time.Sleep(time.Second)
    fmt.Println("do some work1")
}

执行上面代码请点击这里

我们可以发现Go语言中的panic、recover机制，和Java、.NET中的throw、try...catch机制非常类似。

return error的执行机制

return error是利用Go语言函数的多值返回的特性，通过函数的其中一个返回值（一般是第一个或者最后一个），向caller返回函数执行过程中产生的异常，其它值返回执行结果。

这种方式的问题，主要在于：如果函数调用层次比较多，每一层函数都通过return error方式返回错误，都需要处理被调用函数的return error，增加代码复杂度。对于无法恢复的错误也没有必要一层一层往上抛，直接panic/recover更加简洁。

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 (
    "errors"
    "log"
)

type R struct {
}

func f1() (error, *R) {
    return errors.New("an error"), nil
}

func f2() (error, *R) {
    err, r := f1()
    if err != nil {
        return err, nil
    }

    return nil, r
}

func f3() (error, *R) {
    err, r := f2()
    if err != nil {
        return err, nil
    }

    return nil, r
}

func main() {
    err, _ := f3()
    if err != nil {
        log.Print(err)
    }
}

执行上面代码请点击这里

dlog错误处理原则

使用panic还是return error的方式处理错误，要区分不同的场景。重要是不论使用panic还是return error，都需要符合架构上更高层面错误处理需求。

dlog是一个日志记录package，暴露给其它程序调用的方法如下：

• func NewLogger(example interface{}, opts *Options) (*Logger, error)
• func (l *Logger) Log(msg interface{}) error

这两个方法的使用场景并不一样，错误处理原则也不完全一致：

• NewLogger方法一般是在程序初始化的时候调用，用于创建记录程序运行过程中产生的data log的记录器。通过NewLogger创建一个logger的时候，如果传入参数不正确，使用panic方式，在上层调用程序不处理错误情况下会导致程序崩溃，所以使用return error方式向caller报告错误。大多数Golang package也是按此原则处理。
• 上层程序调用logger.Log时，如果Log方法内部发生的错误，不能影响调用的代码的执行，所以这里绝对不能用panic方式抛出错误。日志记录是辅助功能，如果日志记录行为失败，导致业务逻辑代码执行不下去，估计负责业务逻辑开发的工程师会和你拼命。
  • logger.Log可以使用return error方式返回msg校验类的错误
  • logger.Log发送日志采用的是异步批量方式向AWS Kinesis发送数据，向AWS Kinesis发送数据相关的错误无法通过panic或者return error方式直接报告给调用程序。最好的方式是允许调用程序向logger注册发送失败处理的handler，出现发送失败错误时执行handler逻辑。（暂未实现）

如何实现一个logger只能接收对应类型的data log

要实现一个logger只能接收对应类型的data log，主要思路如下：

• Logger的定义中通过属性msgType reflect.Type记住能够接受的消息类型
• 通过NewLogger方法创建logger的时候，指定logger可以接受的消息类型
• Log方法中首先校验msg的类型是否是创建logger时指定的类型

以下是相关代码：

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

// msgType保存Logger能够接受的消息类型
type Logger struct {
    ...
    msgType    reflect.Type
    ...
}

// 获得msg的reflect.Type
func msgType(msg interface{}) (reflect.Type, error) {
    t := reflect.TypeOf(msg)

    if t.Kind() == reflect.Ptr {
        t = t.Elem()
    }

    if t.Kind() != reflect.Struct {
        return nil, fmt.Errorf("dlog message must be either *struct or struct")
    }

    return t, nil
}

func NewLogger(example interface{}, opts *Options) (*Logger, error) {
    t, e := msgType(example)
    if e != nil {
        return nil, e
    }

    ...

    l := &Logger{
        ...
        msgType:    t,
        ...
    }

    ...
    return l, nil
}

func (l *Logger) Log(msg interface{}) error {
    if t, e := msgType(msg); e != nil {
        return e
    } else if !t.AssignableTo(l.msgType) {
        return fmt.Errorf("parameter (%+v) not assignable to %v", msg, l.msgType)
    }

    ...
}

Log方法中为什么要用AssignableTo，而不是直接判断两个类型相等。其实都可以，在msg是struct情况下，AssignableTo返回True意味着两个类型相等。参考下面的例子：

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

package main

import (
    "log"
    "reflect"
)

func main() {
    type Fn func(int) int
    id := func(x int) int {
        return x
    }
    var zeroFn Fn
    log.Println(reflect.TypeOf(id).AssignableTo(reflect.TypeOf(zeroFn)))

    type MyInt int
    mi := 1
    log.Println(reflect.TypeOf(2).AssignableTo(reflect.TypeOf(mi)))

    type S1 struct {
        name string
    }
    type S2 S1

    s1 := S1{
        name: "ethan",
    }
    s2 := S2{
        name: "ethan",
    }
    // s2 = s1	// if uncomment this line, will report "cannot use s1 (type S1) as type S2 in assignment" when compile
    log.Println(reflect.TypeOf(s1).AssignableTo(reflect.TypeOf(s2)))
}

执行上面代码请点击这里

如何实现批量发送data log

要实现批量发送，首先我们可以想到应该要有个buffer用来收集一定数量的的message，等待buffer中的数据积累到一定程度后，一次性发送给AWS Kinesis。设计buffer结构不难，难点在于如何解决多线程(goroutine)并发读写buffer的问题，主要的解决方案有两种：

• 基于锁机制实现对buffer访问控制
• 基于channel实现对buffer的访问控制

前者对于有Java、.NET等语言的并发编程经验的工程师来说，非常熟悉。而后者则体现了CSP(Communicating Sequential Processes)并发编程模型的优势。

dlog的Log方法把收到的msg写到名字叫buffer的channel中，另外一个单独的goroutine在channel的另一头收集编码后的日志信息，然后保存到buf := make([][]byte, 0)中。当buf中的数据量要达到一次向AWS Kinesis发送的最大量时，调用flush方法向AWS Kinesis发送数据。由于只有一个goroutine对buf进行访问，所以不需要通过锁机制控制对buf的读写。

具体代码实现：

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

func NewLogger(example interface{}, opts *Options) (*Logger, error) {
    ...

    go l.sync()    // 启动sync goroutine
    return l, nil
}

func (l *Logger) Log(msg interface{}) error {
    ...

    en := encode(msg)       // 对msg进行编码
    ...
        select {
        case l.buffer <- en:    // 向buffer channel写入编码后的msg
        ...
        }
    ...
    return nil
}

func (l *Logger) sync() {
    ...

    buf := make([][]byte, 0) // 用于收集从buffer channel读取的日志数据
    bufSize := 0

    for {
        select {
        case msg := <-l.buffer:
            if bufSize+len(msg)+partitionKeySize >= maxBatchSize {  // 如果buf的大小接近一次批量发送的最大数据量
                l.flush(&buf, &bufSize)                             // 向AWS Kinesis批量发送数据
            }

            buf = append(buf, msg)                                  // 将从buffer channel读取日志数据保存到buf中
            bufSize += len(msg) + partitionKeySize

        ...
    }
}

如何实现对Logger.Log方法的调用超时机制

如果一个IO操作耗时较长，并且调用比较频繁的情况下，不仅会阻塞caller的执行，还会消耗大量系统资源。我们通常会使用超时机制，避免程序长时间等待或者对系统资源大量占用。

Logger.Log方法利用Go语言channel非常简洁的实现了超时机制：

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

func (l *Logger) Log(msg interface{}) error {
    ...

    var timeout <-chan time.Time
    if l.WriteTimeout > 0 {
        timeout = time.After(l.WriteTimeout)    // 初始化时长为l.WriteTimeout的计时器
    }

    ...
        select {
        case l.buffer <- en:
        case <-timeout: // 如果上一行代码一直阻塞，timeout计时器时间到点后会触发执行当前case下的代码
            return fmt.Errorf("dlog writes %+v timeout after %v", msg, l.WriteTimeout)
        }
    ...
    return nil
}

对比Java、.NET语言中超时机制的实现方法，Go语言的实现简洁的令人发指：

• C#
  • Implementing .Net method timeout
  • How to implement Task Async for a timer in C#?
  • Implementing a timeout in c#
• Java
  • How to implement timeout using threads
  • How to timeout a thread

如何在logger没有收到新msg情况下，保证buf中的数据依然会定期发送给AWS Kinesis

dlog在Logger.sync()方法中通过一个定时器，定期将buf中数据发送给AWS Kinesis。

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

func (l *Logger) sync() {
    if l.SyncPeriod <= 0 {
        l.SyncPeriod = time.Second
    }
    ticker := time.NewTicker(l.SyncPeriod)  // l.SyncPeriod是定期发送的数据的时间间隔，ticker定时触发器

    buf := make([][]byte, 0)
    bufSize := 0

    for {  // 无限循环保证sync goroutine一直工作
        select {
        case msg := <-l.buffer:
            ...

        case <-ticker.C: // ticker.C的类型是<-chan Time，每隔l.SyncPeriod时间会触发执行当前case的代码
            if bufSize > 0 {
                l.flush(&buf, &bufSize)
            }
        }
    }
}

通过ticker，dlog保证了即使没有收到新的msg的时候，保存在buf中的数据最长l.SyncPeriod时间后也会发送给AWS Kinesis。

互联网产品的生产环境的上线，通常的做法是，将现有服务分组，然后交替切流量、升级。如果没有类似的机制，那么在服务程序断掉流量，没有收到新的访问时候，保存在内存中的数据就不会发送出去，升级时就可能导致数据丢失。

如何向程序外部暴露运行指标

Go语言的官方Package expvar提供一种标准化的接口，允许程序暴露公开访问的变量。expvar通过HTTP地址/debug/vars提供访问入口，并以JSON格式展示这些变量。下面是关于expvar常见用法的一个例子：

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

package main

import (
    "encoding/json"
    "expvar"
    "fmt"
    "net/http"
    "sync"
    "time"
)

// Stats is used to collect runtime metrics
type Stats struct {
    sync.Mutex
    TotalHit  int
    ErrorNums int
}

func (s *Stats) IncreaseTotalHit(i int) {
    s.Lock()
    defer s.Unlock()

    s.TotalHit += i
}

func (s *Stats) IncreaseErrorNums(i int) {
    s.Lock()
    defer s.Unlock()

    s.ErrorNums += i
}

func (s *Stats) String() string {
    s.Lock()
    defer s.Unlock()

    b, err := json.Marshal(*s)
    if err != nil {
        return "{}"
    } else {
        return string(b)
    }
}

var (
    stats *Stats
    hits  *expvar.Map
)

func init() {

    expvar.Publish("now", expvar.Func(func() interface{} {
        return time.Now().Format("\"2006-01-02 15:04:05\"")
    }))

    stats = &Stats{}
    expvar.Publish("stats", stats)

    hits = expvar.NewMap("hits").Init()
}

func homeHandler(w http.ResponseWriter, r *http.Request) {
    p := r.URL.Path[1:]
    hits.Add(p, 1)
    stats.IncreaseTotalHit(1)
    fmt.Fprintf(w, "Hey! I love %s! hits: %v\n", p, hits.Get(p))
}

func errHandler(w http.ResponseWriter, r *http.Request) {
    stats.IncreaseErrorNums(1)
    fmt.Fprintf(w, "Error Nums: %v\n", stats.ErrorNums)
}

func main() {
    http.HandleFunc("/err", errHandler)
    http.HandleFunc("/", homeHandler)
    http.ListenAndServe(":8080", nil)
}

按照如下步骤测试运行效果：

• go run expvarexample.go运行例子代码
• 在浏览器中访问http://localhost:8080/ethan
• 在浏览器中访问http://localhost:8080/err
• 在浏览器中访问http://localhost:8080/debug/vars，得到如下结果：

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

  { "cmdline": ["/var/folders/jf/65ft181j33j_d75ktgv67bsc0000gn/T/go-build467453980/command-line-arguments/_obj/exe/expvarsample"], "hits": { "ethan": 1, "favicon.ico": 2 }, "memstats": { ... }, "now": "\"2016-04-19 20:17:40\"", "stats": { "TotalHit":3, "ErrorNums":1 } }

expvarmon是一个帮助查看expvar暴露运行指标的工具，用法如下：

• 安装：go get github.com/divan/expvarmon
• 运行：expvarmon -ports="8080" -vars="hits.ethan,stats.TotalHit,stats.ErrorNums,now"
• 效果如下：

dlog使用expvar向程序外部（比如监控程序）暴露运行指标，目前dlog中定义的运行指标包括：

• writtenRecords: 成功写到AWS Kinesis的msg数量
• writtenBatches: 成功调用AWS Kinesis批量写数据API的次数
• failedRecords: 写到AWS Kinesis失败的msg数量
• tooBigMesssages: 编码后体积过大(加上partitionKeySize大于1MB)的msg数量

未来还需要根据运维的需求对运行指标进行调整，当前的用法也有一些问题，后期需要重构。

如何在单元测试中实现Setup和TearDown

Go语言提供一种轻量级的单元测试框架（无需第三方工具或者程序包）。通过使用go test命令和testing package，可以非常快速的实现单元测试。先借用官方文档中的例子回顾一下Go单元测试框架的用法：

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

//$GOPATH/src/github.com/user/stringutil/reverse_test.go
package stringutil

import "testing"

func TestReverse(t *testing.T) {
    cases := []struct {
        in, want string
    }{
        {"Hello, world", "dlrow ,olleH"},
        {"Hello, 世界", "界世 ,olleH"},
        {"", ""},
    }
    for _, c := range cases {
        got := Reverse(c.in)
        if got != c.want {
            t.Errorf("Reverse(%q) == %q, want %q", c.in, got, c.want)
        }
    }
}

运行测试只需要简单的输入命令：

1
2

$ go test github.com/user/stringutil
ok  	github.com/user/stringutil 0.165s

很多情况下，要执行单元测试，我们需要依赖一些外部资源，比如已完成初始化数据的数据库、公有云上的一些IaaS服务等。这些依赖资源，我们希望在单元测试执行前，能够自动的被初始化；单元测试完成后，能够自动的被清理。testify/suite package就提供这样的支持。通过testify/suite，你可以构建一个测试集struct，建立测试集的setup(初始化)/teardown(清理)方法，和最终实现测试用例逻辑的方法。而运行测试，仍然只需要一句简单的go test。

以下是使用testify/suite实现测试集的常见模式：

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

package suite

import (
    "testing"

    "github.com/stretchr/testify/assert"
)

type SuiteTester struct {
    // Include our basic suite logic.
    Suite

    // Other properties
    propertyN string
}

// The SetupSuite method will be run by testify once, at the very
// start of the testing suite, before any tests are run.
func (suite *SuiteTester) SetupSuite() {
    // ...
}

// The TearDownSuite method will be run by testify once, at the very
// end of the testing suite, after all tests have been run.
func (suite *SuiteTester) TearDownSuite() {
    // ...
}

// The SetupTest method will be run before every test in the suite.
func (suite *SuiteTester) SetupTest() {
    // ...
}

// The TearDownTest method will be run after every test in the suite.
func (suite *SuiteTester) TearDownTest() {
    // ...
}

// a test method
func (suite *SuiteTester) TestOne() {
    // ...
}

// another test method
func (suite *SuiteTester) TestTwo() {
    // ...
}

// TestRunSuite will be run by the 'go test' command, so within it, we
// can run our suite using the Run(*testing.T, TestingSuite) function.
func TestRunSuite(t *testing.T) {
    suiteTester := new(SuiteTester)
    Run(t, suiteTester)
}

dlog中为了测试Logger.Log方法能否正常工作，按照上面的模式编写了相应的测试代码：

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

package dlog

//...

type WriteLogSuiteTester struct {
    suite.Suite

    options     *Options
    seachLogger *Logger
    clickLogger *Logger
    streamNames []string // save the created AWS Kinesis Streams, which will be removed in TearDownSuite()
}

// The SetupSuite method will be run by testify once, at the very
// start of the testing suite, before any tests are run.
func (s *WriteLogSuiteTester) SetupSuite() {

    //...

    // create stream 1
    err = s.seachLogger.kinesis.CreateStream(s.seachLogger.streamName, testingShardCount)
    s.Nil(err)

    // create stream 2
    err = s.clickLogger.kinesis.CreateStream(s.clickLogger.streamName, testingShardCount)
    s.Nil(err)

    s.streamNames = []string{s.seachLogger.streamName, s.clickLogger.streamName}

    for { // waiting created stream's status to be active
        time.Sleep(1 * time.Second)
        resp1, err1 := s.seachLogger.kinesis.DescribeStream(s.seachLogger.streamName)
        s.Nil(err1)

        resp2, err2 := s.seachLogger.kinesis.DescribeStream(s.clickLogger.streamName)
        s.Nil(err2)

        status1 := strings.ToLower(string(resp1.StreamStatus))
        status2 := strings.ToLower(string(resp2.StreamStatus))
        if status1 == "active" && status2 == "active" {
            break
        }
    }
}

// The TearDownSuite method will be run by testify once, at the very
// end of the testing suite, after all tests have been run.
func (s *WriteLogSuiteTester) TearDownSuite() {
    if s.streamNames == nil || len(s.streamNames) == 0 {
        return
    }

    for _, streamName := range s.streamNames {
        err := s.seachLogger.kinesis.DeleteStream(streamName)
        s.Nil(err)
    }
}

func (s *WriteLogSuiteTester) TestWriteLog() {
    defer func() { // Recover if panicking to make sure TearDownSuite will be executed
        if r := recover(); r != nil {
            s.Fail(fmt.Sprint(r))
        }
    }()

    //...
}

func TestRunWriteLogSuite(t *testing.T) {
    suiteTester := new(WriteLogSuiteTester)
    suite.Run(t, suiteTester)
}

注：

• 很多场景下，测试程序自动创建依赖的资源需要运维部门的授权，所以实现前有必要先和运维部门沟通。
• 云环境下，出于安全上的考虑，需要对创建、删除测试资源的账户管理严格管理
  • 账户信息不能写在可以公开访问的测试代码、配置文件中
  • 只给账户分配必要资源的最小权限
  • 为账户能够创建的资源设定配额

如何实现kinesisMock

上一节我们提到在测试执行前初始化依赖资源，现实场景中，并不是任何情况下都能够获得依赖的测试资源，或者测试资源也会出现不可用的情况。通过Mock技术，可以减少测试代码对其它资源（或模块）的依赖。

dlog的测试代码中，首先定义了一个KinesisInterface:

1
2
3
4
5
6

type KinesisInterface interface {
    PutRecords(streamName string, records []kinesis.PutRecordsRequestEntry) (resp *kinesis.PutRecordsResponse, err error)
    CreateStream(name string, shardCount int) error
    DescribeStream(name string) (resp *kinesis.StreamDescription, err error)
    DeleteStream(name string) error
}

KinesisInterface包含了dlog用到的github.com/AdRoll/goamz/kinesis/kinesis的所有方法。因为Go语言interface实现非侵入式的特点，github.com/AdRoll/goamz/kinesis/kinesis自动实现了KinesisInterface，我们再定义一个kinesisMock实现KinesisInterface：

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

type kinesisMock struct {
    // Mapping from steam name to batches of batches
    storage map[string][][]kinesis.PutRecordsRequestEntry

    // simulate lantency that sync to Kinesis
    putRecordLatency time.Duration

    // created streams' names
    streamNames []string

    // lock to solve concurrent call
    lock sync.RWMutex
}

func newKinesisMock(putRecordsLatency time.Duration) *kinesisMock {
    return &kinesisMock{
        storage:          make(map[string][][]kinesis.PutRecordsRequestEntry),
        putRecordLatency: putRecordsLatency,
        streamNames:      make([]string, 0),
    }
}

func (mock *kinesisMock) PutRecords(streamName string, records []kinesis.PutRecordsRequestEntry) (resp *kinesis.PutRecordsResponse, err error) {
    // ...
}

func (mock *kinesisMock) CreateStream(name string, shardCount int) error {
    // ...
}

func (mock *kinesisMock) DescribeStream(name string) (resp *kinesis.StreamDescription, err error) {
    // ...
}

func (mock *kinesisMock) DeleteStream(name string) error {
    // ...
}

然后，把业务代码中所有类型kinesis的变量，替换成KinesisInterface类型。

1
2
3
4
5

type Logger struct {
    //...
    kinesis    KinesisInterface
    //...
}

测试代码中，在构造Logger时传入kinesisMock，而不是真实的kinesis，这样就做到了“狸猫换太子”。

1
2
3
4
5
6
7
8
9
10
11

func TestLoggingToMockKinesis(t *testing.T) {
    assert := assert.New(t)

    l, e := NewLogger(&impression{}, &Options{
        // ...
        UseMockKinesis: true,
        MockKinesis:    newKinesisMock(0),
    })

    // ...
}

如何模拟AWS Kinesis响应慢或者不可用

kinesisMock完全是我们“虚构”出来的一个kinesis，在它的基础上，我们完全可以模拟响应慢或者不可用的情况。

上一节中，不知道大家注意到没有，kinesisMock有个属性叫putRecordLatency，用来模拟调用PutRecords方法的延迟时间。

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

type kinesisMock struct {
    // ...

    // simulate lantency that sync to Kinesis
    putRecordLatency time.Duration

    // ...
}

func (mock *kinesisMock) PutRecords(streamName string, records []kinesis.PutRecordsRequestEntry) (resp *kinesis.PutRecordsResponse, err error) {
    //...

    time.Sleep(mock.putRecordLatency) // 模拟延迟

    //...
}

模拟不可用的kinesis则重新定义了一个brokenKinesisMock：

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

type brokenKinesisMock struct {
    *kinesisMock
}

func newBrokenKinesisMock() *brokenKinesisMock {
    return &brokenKinesisMock{
        kinesisMock: newKinesisMock(0),
    }
}

func (mock *brokenKinesisMock) PutRecords(streamName string, records []kinesis.PutRecordsRequestEntry) (resp *kinesis.PutRecordsResponse, err error) {
    return nil, fmt.Errorf("Kinesis is broken")
}

kinesisMock是brokenKinesisMock的嵌入struct，brokenKinesisMock会自动拥有kinesisMock的所有公开方法，这样也就实现了KinesisInterface。

提交到代码库中的测试代码是否可以保留log.Print

结论是“不可以”，原因总结如下：

• 测试代码中的log.Print，一般用于调试代码，或者在stdout打印出一些信息帮助判断测试失败原因。不论哪种目的，这样的代码目的都仅仅是为了辅助开发，而不应该出现在最终交付的产品代码中。
• go test命令会在控制台输出失败的测试方法，如果加上-v标志会打印出所有测试方法的执行结果，log.Print会影响执行结果的展示效果。团队合作开发，如果每个人都在测试代码中加上自己的log.Print，那么控制台打印出来的测试结果就没法看了。

踩过的一些坑

• AWS Kinesis API - CreateStream是异步创建Stream，而且耗时10+秒，才能完成一个Stream的创建。开始以为是同步创建，结果执行测试逻辑的时候总是出错。
• github.com/AdRoll/goamz/aws/regions.go中缺少中国区AWS Kinesis的URL地址，调用中国区AWS Kinesis会出错。
• Travis CI会Kill掉执行时间超过1分钟的CI过程，而不是如它文档中介绍的“10分钟”

未来可以优化的地方

• 发送失败的错误事件机制
• 实现Kinesis服务不可用或者响应慢的场景下dlog的容错处理

参考

• Effective Go
• Amazon Kinesis Documentation
• Advanced Go Concurrency Patterns
• hystrix-go

招聘消息

我所在的奥阁门科技有限公司正在招聘后端、运维工程师，想加入的朋友、或者有朋友可以推荐的都可以联系我(ethancai@qq.com)。

后端工程师 / Backend Engineer

职责

• 研讨和设计产品功能特性；
• 设计研发系统后端的一个或多个独立服务（micro-service）模块；
• 设计研发业务运营管理系统；
• Code Review。

要求

• 有良好的编程习惯和代码风格；
• 精通至少一种后台开发语言，包括但不限于Go、Node.js、C++、Python；
• 对RESTful、RPC等架构有深刻理解和运用经验；
• 有丰富的web service、web app开发经验；使用过著名的开源应用框架，并完整阅读过源代码；
• 对Mysql、Redis、MongoDB或同类数据存储技术有丰富的使用经验；
• 有提交代码到著名开源库或创建过开源项目者优先；
• 能熟练查阅英文技术文档；
• 有开放、坦诚的沟通心态，乐于分享；
• 5年以上工作经验，3年以上后台系统开发经验。

高级系统运维工程师 / Senior Ops Engineer

职责

• 负责日常业务系统基础实施（AWS）、网络及各子系统的管理维护。
• 负责设计并部署相关应用平台，并提出平台的实施、运行报告。
• 负责配合开发搭建测试平台，协助开发设计、推行、实施和持续改进。
• 负责相关故障、疑难问题排查处理，编制汇总故障、问题，定期提交汇总报告。
• 负责网络监控和应急反应，以确保网络系统有7*24小时的持续运作能力。
• 负责日常系统维护，及监控，提供IT方面的服务和支持，保证系统的稳定。

要求

• 深入理解Linux/Unix操作系统并能熟练使用，了解Linux系统内核，有相关操作系统调优经验优先；
• 熟悉计算机网络基础知识，了解TCP/IP、HTTP等网络协议；
• 熟悉系统服务的管理和维护，例如：Nginx、DNS服务器、NTP服务等；
• 熟悉一种或者多种脚本语言，例如：Shell、Python、Perl 、Ruby等；
• 熟练掌握Linux管理相关命令行工具，例如：grep、awk、sed、tmux、vim等；
• 对数据库系统（MySQL）运维管理有一定的了解；
• 熟悉常见分布式系统系统架构部署管理，熟悉基础设施管理、并具有较强的故障排查和解决问题的能力；
• 具有 2 年以上中大型互联网系统或亚马逊AWS管理经验者优先；
• 有DevOps经验者优先；
• 学习能力和沟通能力较强，具有良好的团队协作精神；
• 工作中需要胆大心细，具备探索创新精神；
• 具有良好的文档编写能力；
• 具有一定的英文技术文档阅读能力。

“体育运动有规则，而对格斗来说，是没有规则的。所以为了迎接格斗，你应该训练身体的所有部位。”
——李小龙

前言

我参加工作是2006年，去的第一家公司是四维图新。那时在技术部从事软件开发工作，功能需求是由技术部的需求组提供的。业务部门要做什么东西，会先给需求组提需求，然后需求组将业务需求整理好，以需求单的形式提供给各个系统开发组。那时候，大家开会说“产品”，指的是公司出版的地图数据，而不是技术部门开发的各个系统、软件工具。学习的内容也是CMMI（软件能力成熟度模型）、RUP（统一软件开发过程）、MSF（微软解决方案框架结构）、UML。

2011年我加入艺龙旅行网，那时艺龙有这么两个部门，一个是网站产品组，一个是业务流程改进组。网站产品负责给面向消费者的预订网站提需求，而业务流程改进组负责给内部呼叫中心使用的系统、面向酒店的业务系统提需求。这两个组后来合并成一个产品团队。我在艺龙负责Ebooking产品的开发和运维工作，和产品经理打了3年的交道，期间自己学习了产品经理的相关知识并且担任一段时间的产品经理。

在艺龙期间以及后来的工作中，我看到一些团队的产品经理很优秀，开发人员技术也很强，但是有时候一个小的需求变更导致技术上的大调整，整个产品开发过程“鸡飞狗跳”，反倒有时候不如以前和一些经验丰富的需求分析师合作时候，产品开发过程更顺畅。

“产品设计”和“需求分析”比较

在传统软件开发企业和互联网企业都工作过的软件工程师，会发现“传统”需求分析师、“互联网”产品经理这两种角色工作内容有着一些明显的区别。

我们通过下面一个测试来了解一下大家所认为的这两个角色的工作内容。下面给出一些概念，如果你认为更接近产品设计，就在产品设计列打“x”，反之在需求分析列下打“x”。

以上概念基本可以分为“关注点”、“考虑问题的模式和方法”、“工作内容”、“常用工具”、“工作交付物”这么几类。相信大部分人选择的接近下面这个答案：

不知道大家做完这个测试后，是不是觉得做产品设计比做需求分析牛逼多了。很多产品经理的书籍中都会将“产品设计”描述的高大上：

• 产品设计是用户价值驱动，需求分析是业务需求驱动。
• 所谓设计，就是通过创造与交流来认识我们生活在其中的世界。好的认识和发现，会让我们感到喜悦和骄傲。
• 互联网产品设计绝不是编写产品交互说明书，也不是创造优美的界面，本质是创造一连串的体验，使用户能感知到产品的文化、价值和内涵，从而引发集群效应、创造社会价值。

以上的话说的都对，绝对正确。在乔老爷子的引领下，如今很多公司重视产品设计，这是好事。但是于此同时产生另外一种做法：不重视需求分析，甚至直接忽略。老板、产品经理这么做可以理解，工程师如果也这么附和，就是做事不过大脑。

需求分析是否有继续存在的价值

产品设计要取代需求分析，只可能是下面两种情况：

• 产品设计涵盖了需求分析的所有工作内容
• 属于需求分析但是不属于产品设计的工作内容已不再重要，可以不用去做

第一种情况，需求分析中的业务建模、画E-R关系图、UML流程图、时序图、状态图，在大多数公司中，是由需求分析师或者工程师来做，不属于产品设计的工作范畴。产品设计显然不涵盖需求分析的所有工作内容。

第二种情况，面对复杂业务场景，通过业务建模、画E-R关系图、UML流程图、时序图、状态图等手段做需求分析，是前辈软件工程师总结出来宝贵方法和经验。产品经理一般会将整理后的产品需求通过各种方式（口述、文档、UI/UE设计等）传递给工程团队，但是这些信息一般都是站在业务边界或系统边界描述的产品功能需求，和工程实现存在信息断层。产品需求如何映射到代码实现，需要一个系统化的需求分析、设计过程，产品设计并不解决过去几十年软件开发行业发明各种软件需求分析方法所解决的问题。任何脑袋没有烧掉的公司都不会说软件需求分析工作不重要、不需要做。

需求分析应继续得到坚持和重视

如果只关注产品设计，更多的时候，你会认为设计出来的界面很美观，交互也很舒服，该说的需求都告诉工程团队了，但是工程开发阶段，出现各种问题：概念不明确、架构无法设计、逻辑混乱、后期变更困难。

如果只关注需求分析，更多的时候，你会发现不知道如何提产品需求，产品开发出来了，认为做很牛逼，但是客户觉得不好用，产品不好卖。

所有公司都希望做出成功的产品。但是未经仔细思考，就忽略软件需求分析方法，不是一个经得起推敲的做法。只有产品设计、需求分析都做好了，产品才能成功。

参考

• UMLChina
• 李小龙截拳道