（架构图源于参考书籍）
Elasticsearch 环境搭建官方简介：Elasticsearch 是一个分布式、RESTful 风格的搜索和数据分析引擎，能够解决不断涌现出的各种用例。 作为 Elastic Stack 的核心，它集中存储您的数据，帮助您发现意料之中以及意料之外的情况。
下载安装# 下载 wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-7.15.1-linux-x86_64.tar.gz # 解压 tar xzvf elasticsearch-7.15.1-linux-x86_64.tar.gz # 以非 root 用户启动 cd /elasticsearch-7.15.1/bin/ ./elasticsearch
# 检验是否启动成功，172.16.16.4 为 elasticsearch.yml 配置绑定的 IP 地址 curl 172.16.16.4:9200
若无法正常启动，则修改配置：
/home/sam/elasticsearch-7.15.1/config 修改 jvm.options 中内存配置： -Xms256m -Xmx256m 修改 vim elasticsearch.yml ： cluster.name: my-application node.name: node-1 network.host: 172.16.16.4 http.port: 9200 discovery.seed_hosts: ["172.26.26.4", "::1"] cluster.initial_master_nodes: ["node-1"]
若出现以下报错，则需修改系统配置：
# 报错 bootstrap check failure [1] of [1]: max virtual memory areas vm.max_map_count [65530] is too low, increase to at least [262144] # 解决 sudo vim /etc/sysctl.conf # 添加以下内容 vm.max_map_count=262144 # 重载配置 sudo sysctl -p # 重新启动 sudo systemctl start elasticsearch配置开机启动
# 创建开机启动文件 vim /usr/lib/systemd/system/elasticsearch.service 内容如下： [Unit] Description=elasticsearch [Service] User=sam #启动用户 LimitNOFILE=100000 LimitNPROC=100000 ExecStart=/home/sam/elasticsearch-7.15.1/bin/elasticsearch #安装路径 [Install] WantedBy=multi-user.target # 重新加载文件配置 systemctl daemon-reload # 设置开机启动 systemctl enable elasticsearch # 关掉之前启动的 es lsof -i tcp:9200 kill -9 pid # 启动 es systemctl start elasticsearch开启远程连接
# 放行端口 iptables -I INPUT 4 -p tcp -m state --state NEW -m tcp --dport 9200 -j ACCEPT # 保存 iptables 规则 service iptables save # 远程测试 curl 公网IP:9200elasticsearch-head 插件
# github 地址 https://github.com/mobz/elasticsearch-head # npm 启动方式 git clone git://github.com/mobz/elasticsearch-head.git cd elasticsearch-head npm install npm run start open http://localhost:9100/
# 搭建 node 环境 # 查看当前有那些可供选择的版本 # dnf module list node.js # 选择一个版本 # dnf module edable nodejs:14 # 安装 nodejs dnf install nodejs # 查看当前的版本 node --version npm --version
# 进入目录并安装 cd elasticsearch-head npm install
vim elasticsearch.yml # 配置跨域 http.cors.enabled: true http.cors.allow-origin: "*" # 重启 es 服务 sudo systemctl restart elasticsearch.service
# 放行端口 iptables -I INPUT 4 -p tcp -m state --state NEW -m tcp --dport 9100 -j ACCEPT # 保存 iptables 规则 service iptables save # 启动 npm run start
远程访问：
创建映射参考：官方 API 文档
创建 metadata 索引以及 objects 类型的映射：
curl -H "Content-Type: application/json" -XPUT 172.16.16.4:9200/metadata?include_type_name=true -d'{"mappings":{"objects":{"properties":{"name":{"type":"text","fielddata": true},"version":{"type":"integer"},"size":{"type":"integer"},"hash":{"type":"text"}}}}}'
在页面可以看到索引和集群信息，这里有一个 Unassigned 的节点是因为创建映射时默认副本为 1 ，而我们使用的是单节点，但是此时服务器依然可以使用。
ES包封装该 ES 包封装了以 HTTP 访问 ES 的各种 API 的操作。
package es
import (
"demo/sys"
"encoding/json"
"fmt"
"io/ioutil"
"net/http"
"net/url"
"os"
"strings"
)
type Metadata struct {
Name string
Version int
Size int64
Hash string
}
type hit struct {
Source Metadata `json:"_source"`
}
type searchResult struct {
Hits struct {
Total int
Hits []hit
}
}
func getMetadata(name string, versionId int) (meta Metadata, e error) {
// 索引为 metadata ，类型为 objects，文档 id 为对象名称和版本号的拼接
url := fmt.Sprintf(sys.GetMetadataUrl, os.Getenv(sys.EsServer), name, versionId)
// 通过 GET URL 可以直接获取该对象的元数据，免除了耗时的搜索操作
r, e := http.Get(url)
if e != nil {
return
}
if r.StatusCode != http.StatusOK {
e = fmt.Errorf(sys.FailToGetMetadata, name, versionId, r.StatusCode)
return
}
result, _ := ioutil.ReadAll(r.Body)
// 将请求结果反序列化为元数据结构
json.Unmarshal(result, &meta)
return
}
func SearchLatestVersion(name string) (meta Metadata, e error) {
// 构建 url 时需要将名称转移成 url 字符
url := fmt.Sprintf(sys.SearchLatestVersionUrl, os.Getenv(sys.EsServer), url.PathEscape(name))
r, e := http.Get(url)
if e != nil {
return
}
if r.StatusCode != http.StatusOK {
e = fmt.Errorf(sys.FailToSearchLatestMetadata, r.StatusCode)
return
}
result, _ := ioutil.ReadAll(r.Body)
var sr searchResult
// 请求结果反序列化
json.Unmarshal(result, &sr)
// 如果长度为 0 则没有搜索结果，直接返回
if len(sr.Hits.Hits) != 0 {
meta = sr.Hits.Hits[0].Source
}
return
}
func GetMetadata(name string, version int) (Metadata, error) {
// 没有指定版本号时默认返回最新版本的元数据
if version == 0 {
return SearchLatestVersion(name)
}
return getMetadata(name, version)
}
func PutMetadata(name string, version int, size int64, hash string) error {
doc := fmt.Sprintf(sys.MetadataJson, name, version, size, hash)
client := http.Client{}
url := fmt.Sprintf(sys.PutMetadataUrl, os.Getenv(sys.EsServer), name, version)
request, _ := http.NewRequest(http.MethodPut, url, strings.NewReader(doc))
// 加入 header ，否则报 406
request.Header.Add("content-type","application/json")
r, e := client.Do(request)
if e != nil {
return e
}
if r.StatusCode == http.StatusConflict {
return PutMetadata(name, version+1, size, hash)
}
if r.StatusCode != http.StatusCreated {
result, _ := ioutil.ReadAll(r.Body)
return fmt.Errorf(sys.FailToPutMetadata, r.StatusCode, string(result))
}
return nil
}
func AddVersion(name, hash string, size int64) error {
// 获取目前最新的版本
version, e := SearchLatestVersion(name)
if e != nil {
return e
}
// 创建一个最新的版本号
return PutMetadata(name, version.Version+1, size, hash)
}
func SearchAllVersions(name string, from, size int) ([]Metadata, error) {
// 不指定名字时则搜索全部对象的全部版本，指定名字时则搜索某个对象的全部版本
url := fmt.Sprintf(sys.SearchAllVersionsUrl, os.Getenv(sys.EsServer), from, size)
if name != "" {
url += "&q=name:" + name
}
r, e := http.Get(url)
if e != nil {
return nil, e
}
metas := make([]Metadata, 0)
result, _ := ioutil.ReadAll(r.Body)
var sr searchResult
json.Unmarshal(result, &sr)
for i := range sr.Hits.Hits {
metas = append(metas, sr.Hits.Hits[i].Source)
}
return metas, nil
}
func DelMetadata(name string, version int) {
client := http.Client{}
url := fmt.Sprintf(sys.DelMetadataUrl, os.Getenv(sys.EsServer), name, version)
request, _ := http.NewRequest(http.MethodDelete, url, nil)
client.Do(request)
}
type Bucket struct {
Key string
Doc_count int
Min_version struct {
Value float32
}
}
type aggregateResult struct {
Aggregations struct {
Group_by_name struct {
Buckets []Bucket
}
}
}
func SearchVersionStatus(min_doc_count int) ([]Bucket, error) {
client := http.Client{}
url := fmt.Sprintf(sys.SearchVersionStatusUrl, os.Getenv(sys.EsServer))
body := fmt.Sprintf(sys.SearchVersionStatusJson, min_doc_count)
request, _ := http.NewRequest(http.MethodGet, url, strings.NewReader(body))
r, e := client.Do(request)
if e != nil {
return nil, e
}
b, _ := ioutil.ReadAll(r.Body)
var ar aggregateResult
json.Unmarshal(b, &ar)
return ar.Aggregations.Group_by_name.Buckets, nil
}
func HasHash(hash string) (bool, error) {
url := fmt.Sprintf(sys.HasHashUrl, os.Getenv(sys.EsServer), hash)
r, e := http.Get(url)
if e != nil {
return false, e
}
b, _ := ioutil.ReadAll(r.Body)
var sr searchResult
json.Unmarshal(b, &sr)
return sr.Hits.Total != 0, nil
}
func SearchHashSize(hash string) (size int64, e error) {
url := fmt.Sprintf(sys.SearchHashSizeUrl, os.Getenv(sys.EsServer), hash)
r, e := http.Get(url)
if e != nil {
return
}
if r.StatusCode != http.StatusOK {
e = fmt.Errorf(sys.FailToSearchHashSize, r.StatusCode)
return
}
result, _ := ioutil.ReadAll(r.Body)
var sr searchResult
json.Unmarshal(result, &sr)
if len(sr.Hits.Hits) != 0 {
size = sr.Hits.Hits[0].Source.Size
}
return
}
版本信息搜索
package version
import (
"demo/es"
"encoding/json"
"log"
"net/http"
"strings"
)
func Handler(w http.ResponseWriter, r *http.Request) {
// 非 GET 方法时响应方法不允许
m := r.Method
if m != http.MethodGet {
w.WriteHeader(http.StatusMethodNotAllowed)
return
}
// 其实是分页参数，一页最多有 1000 条记录，默认从第 0 条开始往后取数据
// 当返回值的长度不等于 1000 时，则说明后续没有数据了，直接返回
// 当返回值等于 1000 时，说明后续可能有数据， from 则从 1000 条开始往后取数据
from := 0
size := 1000
// 若未指定名字，则切割 URL 之后名字为空字符串
name := strings.Split(r.URL.EscapedPath(), "/")[2]
for {
metas, e := es.SearchAllVersions(name, from, size)
if e != nil {
log.Println(e)
// 服务器内部错误
w.WriteHeader(http.StatusInternalServerError)
return
}
// 遍历结果集
for i := range metas {
// 格式化为 json 返回
b, _ := json.Marshal(metas[i])
w.Write(b)
w.Write([]byte("n"))
}
if len(metas) != size {
return
}
from += size
}
}
散列（hash）工具类封装
package utils
import (
"crypto/sha256"
"encoding/base64"
"io"
"net/http"
"strconv"
"strings"
)
func GetOffsetFromHeader(h http.Header) int64 {
byteRange := h.Get("range")
if len(byteRange) < 7 {
return 0
}
if byteRange[:6] != "bytes=" {
return 0
}
bytePos := strings.Split(byteRange[6:], "-")
offset, _ := strconv.ParseInt(bytePos[0], 0, 64)
return offset
}
func GetHashFromHeader(h http.Header) string {
digest := h.Get("digest")
if len(digest) < 9 {
return ""
}
if digest[:8] != "SHA-256=" {
return ""
}
return digest[8:]
}
func GetSizeFromHeader(h http.Header) int64 {
size, _ := strconv.ParseInt(h.Get("content-length"), 0, 64)
return size
}
func CalculateHash(r io.Reader) string {
h := sha256.New()
io.Copy(h, r)
return base64.StdEncoding.EncodeToString(h.Sum(nil))
}
PUT、GET、DELETe 处理函数
package objects
import (
"demo/apiServer/heartbeat"
"demo/apiServer/locate"
"demo/apiServer/objectStream"
"demo/es"
"demo/sys"
"demo/utils"
"fmt"
"io"
"log"
"net/http"
"net/url"
"strconv"
"strings"
)
func Handler(w http.ResponseWriter, r *http.Request) {
m := r.Method
// PUT 方法时，创建或者替换资源
if m == http.MethodPut {
put(w, r)
return
}
// GET 方法时，获取资源
if m == http.MethodGet {
get(w, r)
return
}
// 版本删除
if m == http.MethodDelete {
del(w,r)
return
}
// 其他方式时，返回状态码，方法不允许
w.WriteHeader(http.StatusMethodNotAllowed)
}
func put(w http.ResponseWriter, r *http.Request) {
// 按以前的步骤，这里应该获取存储对象名字，不过从 header 中取对象的散列值作为名字
hash := utils.GetHashFromHeader(r.Header)
if hash == "" {
log.Println(sys.MissingObjectHash)
w.WriteHeader(http.StatusBadRequest)
return
}
// 存储请求数据，散列值要作转义
httpStatus, e := storeObject(r.Body, url.PathEscape(hash))
if e != nil {
log.Println(e)
w.WriteHeader(httpStatus)
return
}
if httpStatus != http.StatusOK {
w.WriteHeader(httpStatus)
return
}
// 获取名字和大小，新增一个对象版本
name := strings.Split(r.URL.EscapedPath(), "/")[2]
size := utils.GetSizeFromHeader(r.Header)
e = es.AddVersion(name, hash, size)
if e != nil {
log.Println(e)
w.WriteHeader(http.StatusInternalServerError)
}
// 返回结果
w.WriteHeader(httpStatus)
}
func storeObject(r io.Reader, obj string) (int, error) {
// 获取接口服务节点存储对象的流
stream, e := putStream(obj)
if e != nil {
return http.StatusServiceUnavailable, e
}
// 将请求数据体拷贝到流 stream
io.Copy(stream, r)
// 关闭流
e = stream.Close()
if e != nil {
return http.StatusInternalServerError, e
}
// 返回成功状态码
return http.StatusOK, nil
}
func putStream(obj string) (*objectStream.PutStream, error) {
// 随机选择一个数据服务节点
server := heartbeat.ChooseRandomDataServer()
// 若没有可用的数据服务节点则返回错误信息
if server == "" {
return nil, fmt.Errorf(sys.DataServerNotFound)
}
// 返回数据服务节点存储对象的流
return objectStream.NewPutStream(server, obj), nil
}
func get(w http.ResponseWriter, r *http.Request) {
// 获取存储对象名称和版本号
name := strings.Split(r.URL.EscapedPath(), "/")[2]
versionId := r.URL.Query()["version"]
version := 0
var e error
if len(versionId) != 0 {
// 版本号字符串转数字
version, e = strconv.Atoi(versionId[0])
if e != nil {
log.Println(e)
w.WriteHeader(http.StatusBadRequest)
return
}
}
// 根据名字和版本号来获取元数据
meta, e := es.GetMetadata(name, version)
if e != nil {
log.Println(e)
w.WriteHeader(http.StatusInternalServerError)
return
}
// 元数据散列值为空则无该对象
if meta.Hash == "" {
w.WriteHeader(http.StatusNotFound)
return
}
// 散列值要作 URL 转移
object := url.PathEscape(meta.Hash)
// 根据散列值获取对象数据
stream, e := getStream(object)
if e != nil {
log.Println(e)
w.WriteHeader(http.StatusNotFound)
return
}
// 将数据流拷贝到响应流 w
io.Copy(w, stream)
}
func getStream(obj string) (io.Reader, error) {
// 根据存储对象名称进行定位
server := locate.Locate(obj)
// 未找到该存储对象时返回定位失败错误
if server == "" {
return nil, fmt.Errorf(sys.DataServerLocateFail, obj)
}
// 定位到存储对象时，返回该对象的数据流
return objectStream.NewGetStream(server, obj)
}
func del(w http.ResponseWriter, r *http.Request) {
// 获取名字
name := strings.Split(r.URL.EscapedPath(),"/")[2]
v,e := es.SearchLatestVersion(name)
if e != nil {
log.Println(e)
w.WriteHeader(http.StatusInternalServerError)
return
}
// 插入一条新的元数据作删除标记
e = es.PutMetadata(name,v.Version + 1,0,"")
if e != nil {
log.Println(e)
w.WriteHeader(http.StatusInternalServerError)
return
}
}
数据服务节点
主要是散列值进行 URL 转义，传参时注意这点，否则找不到文件。如果这里不作处理也行，可以由前端传参时进行转义。若散列值有斜杠则必须先转义，再做 GET 请求，因为取文件名时是切割 URL 取第三个值作为名字。
package locate
import (
"demo/rabbitmq"
"demo/sys"
"log"
"net/url"
"os"
"strconv"
)
// 定位对象
func Locate(name string) bool {
// 访问磁盘上对应的文件名
_, e := os.Stat(name)
// 判读文件名是否存在
return !os.IsNotExist(e)
}
// 监听定位信息
func StartLocate() {
q := rabbitmq.New(os.Getenv(sys.RabbitmqServer))
defer q.Close()
// 绑定 data 网络层
q.Bind(sys.DataServersExchange)
// 获取信息管道
c := q.Consume()
// 从管道中遍历信息，msg 为需要定位的存储对象名字
for msg := range c {
// 去掉 json 序列化的双引号
obj, e := strconv.Unquote(string(msg.Body))
if e != nil {
log.Fatalln(e)
}
// 存储根目录拼接文件名，定位存储对象，名字需要 URL 转义
if Locate(os.Getenv(sys.StorageRoot) + url.PathEscape(obj)) {
// 如果存储对象存在，则回送本节点监听地址，已告知存储对象在该节点
q.Send(msg.ReplyTo, os.Getenv(sys.ListenAddress))
}
}
}
main 函数
package main
import (
"demo/apiServer/heartbeat"
"demo/apiServer/locate"
"demo/apiServer/objects"
"demo/apiServer/version"
"demo/sys"
"net/http"
"os"
)
func main() {
// 监听数据服务节点心跳
go heartbeat.ListenHeartbeat()
// 处理对象请求，实际上是将对象请求转发给数据服务
http.HandleFunc("/handleObjs/", objects.Handler)
// 处理定位请求
http.HandleFunc("/locateObj/", locate.Handler)
// 处理版本信息
http.HandleFunc("/versions/",version.Handler)
// 启动并监听服务
http.ListenAndServe(os.Getenv(sys.ListenAddress), nil)
}
测试
模拟分布式网络
# 查看本机网络接口 ip a # 数据服务节点 eth0:1~6 # IP范围 172.16.17.1 ~ 172.16.17.6 # 接口服务节点 eth0:7~8 # IP范围 172.16.18.1 ~ 172.16.18.2 # 网络接口绑定多个 IP ifconfig eth0:1 172.16.17.1/20 ifconfig eth0:2 172.16.17.2/20 ifconfig eth0:3 172.16.17.3/20 ifconfig eth0:4 172.16.17.4/20 ifconfig eth0:5 172.16.17.5/20 ifconfig eth0:6 172.16.17.6/20 ifconfig eth0:7 172.16.18.1/20 ifconfig eth0:8 172.16.18.2/20导入环境变量
# rabbitmq-server 变量 export RABBITMQ_SERVER=amqp://yushanma:passwd@172.16.16.4:5672 # es-server 变量 export ES_SERVER=172.16.16.4:9200启动服务
# 启动数据服务节点 LISTEN_ADDRESS=172.16.17.1:12345 STORAGE_ROOT=/home/sam/files/1/objects/ go run dataServer/cmd/main.go & LISTEN_ADDRESS=172.16.17.2:12345 STORAGE_ROOT=/home/sam/files/2/objects/ go run dataServer/cmd/main.go & LISTEN_ADDRESS=172.16.17.3:12345 STORAGE_ROOT=/home/sam/files/3/objects/ go run dataServer/cmd/main.go & LISTEN_ADDRESS=172.16.17.4:12345 STORAGE_ROOT=/home/sam/files/4/objects/ go run dataServer/cmd/main.go & LISTEN_ADDRESS=172.16.17.5:12345 STORAGE_ROOT=/home/sam/files/5/objects/ go run dataServer/cmd/main.go & LISTEN_ADDRESS=172.16.17.6:12345 STORAGE_ROOT=/home/sam/files/6/objects/ go run dataServer/cmd/main.go & # 启动接口服务节点 LISTEN_ADDRESS=172.16.18.1:12346 go run apiServer/cmd/main.go & LISTEN_ADDRESS=172.16.18.2:12346 go run apiServer/cmd/main.go &功能测试
选择第一个服务节点 172.16.18.1:12346 ， PUT 一个名为 hello 的对象：
curl -v 172.16.18.1:12346/handleObjs/hello -XPUT -d"hello,yushanma"
因为我们没有提供散列值，因此会报 400 错误。我们可以通过 openssl 计算出这个对象的散列值：
echo -n "hello,yushanma" | openssl dgst -sha256 -binary | base64
将该散列值加入 PUT 请求的 header ：
curl -v 172.16.18.1:12346/handleObjs/hello -XPUT -d"hello,yushanma" -H "digest:SHA-256=Ziupvid3V+SN3UJU3RjhrU3rZqykZPIN1JJCmT31vuo="
OK，我们已经上传了一个 hello 对象，也是第一个版本，接下来换第二个接口服务节点 172.16.18.2:12346 上传 hello 第二个版本：
# 计算散列值 echo -n "hello,yushanma,shirley" | openssl dgst -sha256 -binary | base64 YYmSQFej7Zs/82P2ZFe1NE8dtZQI1NCYOnlRj/EQi6w= curl -v 172.16.18.1:12346/handleObjs/hello -XPUT -d"hello,yushanma,shirley" -H "digest:SHA-256=YYmSQFej7Zs/82P2ZFe1NE8dtZQI1NCYOnlRj/EQi6w="
至此，我们上传了两个 hello 对象，接下来用定位接口去查看它们分别被保存到哪个数据服务节点上：
Jack Ma@DESKTOP-L24D7IP MINGW64 ~/Desktop $ echo -n "hello,yushanma" | openssl dgst -sha256 -binary | base64 Ziupvid3V+SN3UJU3RjhrU3rZqykZPIN1JJCmT31vuo= Jack Ma@DESKTOP-L24D7IP MINGW64 ~/Desktop $ echo -n "hello,yushanma,shirley" | openssl dgst -sha256 -binary | base64 YYmSQFej7Zs/82P2ZFe1NE8dtZQI1NCYOnlRj/EQi6w= curl 172.16.18.1:12346/locateObj/Ziupvid3V+SN3UJU3RjhrU3rZqykZPIN1JJCmT31vuo= curl 172.16.18.2:12346/locateObj/YYmSQFej7Zs%2F82P2ZFe1NE8dtZQI1NCYOnlRj%2FEQi6w=
能在对应的存储位置找到这两个文件。下面查看 hello 文件的版本：
curl 172.16.18.1:12346/versions/hello
可以看到有两个版本的 hello ，我们尝试 GET 请求 hello ：
# 不传参数 version 时默认获取最新版本 curl 172.16.18.2:12346/handleObjs/hello?version=2
可以看到 GET 请求响应符合预期，最后尝试删除 hello ：
curl -v 172.16.18.2:12346/handleObjs/hello -XDELETE
可以看到，删除时是将 hello 的最新版本标记为删除状态，大小为 0 ，没有散列值，再次请求最新的 hello 版本时响应 404 not found ，符合预期。但我们依然可以通过指定版本号来获取存在的 hello 。
