go语言开发限流器插件 go 限流器

coredns源码分析

CoreDNS是使用go语言编写的快速灵活的DNS服务，采用链式插件模式，每个插件实现独立的功能，底层协议可以是tcp/udp，也可以是TLS，gRPC等。默认监听所有ip地址，可使用bind插件指定监听指定地址。

成都创新互联公司服务紧随时代发展步伐，进行技术革新和技术进步，经过10年的发展和积累，已经汇集了一批资深网站策划师、设计师、专业的网站实施团队以及高素质售后服务人员，并且完全形成了一套成熟的业务流程，能够完全依照客户要求对网站进行做网站、成都网站设计、建设、维护、更新和改版，实现客户网站对外宣传展示的首要目的，并为客户企业品牌互联网化提供全面的解决方案。

格式如下

SCHEME是可选的，默认值为dns://，也可以指定为tls://，grpc://或者https://。

ZONE是可选的，指定了此dnsserver可以服务的域名前缀，如果不指定，则默认为root，表示可以接收所有的dns请求。

PORT是选项的，指定了监听端口号，默认为53，如果这里指定了端口号，则不能通过参数-dns.port覆盖。

一块上面格式的配置表示一个dnsserver，称为serverblock，可以配置多个serverblock表示多个dnsserver。

下面通过一个例子说明，如下配置文件指定了4个serverblock，即4个dnsserver，第一个监听端口5300，后面三个监听同一个端口53，每个dnsserver指定了特定的插件。

下图为配置的简略图

a. 从图中可看到插件执行顺序不是配置文件中的顺序，这是因为插件执行顺序是在源码目录中的plugin.cfg指定的，一旦编译后，顺序就固定了。

b. .根serverblock虽然指定了health，但是图中却没有，这是因为health插件不参与dns请求的处理。能处理dns请求的插件必须提供如下两个接口函数。

dns请求处理流程

收到dns请求后，首先根据域名匹配zone找到对应的dnsserver(最长匹配优先)，如果没有匹配到，则使用默认的root dnsserver。

找到dnsserver后，就要按照插件顺序执行其中配置的插件，当然并不是配置的插件都会被执行，如果某个插件成功找到记录，则返回成功，否则根据插件是否配置了fallthrough等来决定是否执行下一个插件。

plugin.cfg

源码目录下的plugin.cfg指定了插件执行顺序，如果想添加插件，可按格式添加到指定位置。

源码目录下的Makefile根据plugin.cfg生成了两个go文件：zplugin.go和zdirectives.go。

core/dnsserver/zdirectives.go将所有插件名字放在一个数组中。

codedns 主函数

codedns.go 首先导入了包"github点抗 /coredns/coredns/core/plugin"，此包内只有一个文件zplugin.go，此文件为自动生成的，主要导入了所有的插件，执行每个插件的init函数。

接着执行 run.go Run

此文件又引入了包"github点抗 /coredns/coredns/core/dnsserver"，其init函数在 dnsserver/register.go 文件中，如下所示，主要是注册了serverType

剩下的就是解析参数，解析配置文件后，执行caddy.Start。

这里就是根据配置文件中指定的serverblock，执行插件的setup进行初始化，创建对应的server，开始监听dns请求

tcp协议调用Serve，udp协议调用ServePacket

收到DNS请求后，调用ServeDNS，根据域名匹配dnsserver，如果没有匹配不到则使用根dnsserver，然后执行dnsserver中配置的插件

以k8s插件为例

参考

//如何写coredns插件

//coredns源码分析

//NodeLocal DNSCache

【GO】golang 降级|熔断|限流实战

做为本文的前言，首先向读者介绍一下降级、熔断和限流的概念与关系。也许很多人对此，早已谙熟于心，但是烦请允许我再啰嗦几句，方便第一次接触该领域的小伙伴们，都可以的理解消化本文。

所谓限流，本质就是对系统的被请求频率以及内部的部分功能的执行频率加以限制，防止因突发的流量激增，导致整个系统不可用。当流量出现激增，触发限流，那么对于那些系统暂时不想或无法处理的“流量”，我们该如何处理呢？这就自然引出了服务降级的概念，其本质就是提供降低系统正常运行所能提供的功能数，亦或是降低某些功能完成的完整度（质量）。而熔断就是众多降级手段中最常见的一种，其在流量过大时（或下游服务出现问题时），可以自动断开与下游服务的交互，并可以通过自我诊断下游系统的错误是否已经修正，或上游流量是否减少至正常水平，来恢复自我恢复。

简而言之，限流是从系统的流量入口考虑，从进入的流量上进行限制，达到保护系统的作用；降级，是从系统内部的平级服务或者业务的维度考虑，流量大了，可以干掉一些，保护其他正常使用；熔断强调的是服务之间的调用能实现自我恢复的状态；

Hystrix的golang版本项目地址是：

Hystrix是Netflix开源的一个限流熔断的项目、主要有以下功能:

项目地址为：

gobreaker是索尼的开源的一个限流熔断的项目，是基于《微软云设计模式》一书中的熔断器模式的 Golang 实现的，本质利用的还是原子计数法、主要有以下功能:

如何在 Go 语言中用 Beats 开发 Logstash 插件

配置环境

在OSX系统上很容易安装GO的可执行文件:

brew install go

虽然Java或Ruby (或者任何我知道的语言) 可以在本地文件系统的任何地方使用命令,，但是Go项目必须使用单一专用的地址,，并且在$GOPATH环境变量下可用。

第 1 段（可获 1.54 积分）

R e; 7个月前 

创建项目对于Logstash插件，Beats项目可以从模板创建。官方文档的说明十分简单。鉴于Go对文件系统上的位置的严格要求，只需按照以下说明生成一个新的即可使用的Go项目。默认模板代码将在控制台中重复发送带增量计数器的事件：./redditbeat -e -d "*"

2016/12/13 22:55:56.013362 beat.go:267: INFO

Home path: [/Users/i303869/projects/private/go/src/github点抗 /nfrankel/redditbeat]

Config path: [/Users/i303869/projects/private/go/src/github点抗 /nfrankel/redditbeat]

Data path: [/Users/i303869/projects/private/go/src/github点抗 /nfrankel/redditbeat/data]

Logs path: [/Users/i303869/projects/private/go/src/github点抗 /nfrankel/redditbeat/logs]

2016/12/13 22:55:56.013390 beat.go:177: INFO Setup Beat: redditbeat; Version: 6.0.0-alpha1

2016/12/13 22:55:56.013402 processor.go:43: DBG  Processors: 

2016/12/13 22:55:56.013413 beat.go:183: DBG  Initializing output plugins

2016/12/13 22:55:56.013417 logp.go:219: INFO Metrics logging every 30s

2016/12/13 22:55:56.013518 output.go:167: INFO Loading template enabled. Reading template file:

/Users/i303869/projects/private/go/src/github点抗 /nfrankel/redditbeat/redditbeat.template.json

2016/12/13 22:55:56.013888 output.go:178: INFO Loading template enabled for Elasticsearch 2.x. Reading template file:

/Users/i303869/projects/private/go/src/github点抗 /nfrankel/redditbeat/redditbeat.template-es2x.json

2016/12/13 22:55:56.014229 client.go:120: INFO Elasticsearch url: 

2016/12/13 22:55:56.014272 outputs.go:106: INFO Activated elasticsearch as output plugin.

2016/12/13 22:55:56.014279 publish.go:234: DBG  Create output worker

2016/12/13 22:55:56.014312 publish.go:276: DBG  No output is defined to store the topology.

The server fields might not be filled.

2016/12/13 22:55:56.014326 publish.go:291: INFO Publisher name: LSNM33795267A

2016/12/13 22:55:56.014386 async.go:63: INFO Flush Interval set to: 1s

2016/12/13 22:55:56.014391 async.go:64: INFO Max Bulk Size set to: 50

2016/12/13 22:55:56.014395 async.go:72: DBG  create bulk processing worker (interval=1s, bulk size=50)

2016/12/13 22:55:56.014449 beat.go:207: INFO redditbeat start running.

2016/12/13 22:55:56.014459 redditbeat.go:38: INFO redditbeat is running! Hit CTRL-C to stop it.

2016/12/13 22:55:57.370781 client.go:184: DBG  Publish: {

"@timestamp": "2016-12-13T22:54:47.252Z",

"beat": {

"hostname": "LSNM33795267A",

"name": "LSNM33795267A",

"version": "6.0.0-alpha1"

},

"counter": 1,

"type": "redditbeat"

}

第 2 段（可获 0.73 积分）

R e; 7个月前 

关于命令行参数：-e记录到标准err，而-d“*”启用所有调试选择器。有关参数的完整列表，请键入./redditbeat --help。编码Go代码位于.go文件中（令人惊讶...）在$ GOPATH / src文件夹的项目子文件夹中。配置类型第一个有趣的文件是config / config.go，它定义了一个结构来声明Beat的可能参数。至于前面的Logstash插件，让我们添加一个subreddit参数，并设置它的默认值：type Config struct {

Period time.Duration `config:"period"`

Subreddit string `config:"subreddit"`

}

var DefaultConfig = Config {

Period: 15 * time.Second,

Subreddit: "elastic",

}

第 3 段（可获 0.89 积分）

R e; 7个月前 

Beater TypeBeat本身的代码在beater / redditbean.go中找到。默认模板为Beat和三个函数创建一个struct：Beat构造函数—用来读取配置: func New(b *beat.Beat, cfg *common.Config) (beat.Beater, error) { ... } 

Run 函数- 需要覆盖Beat的主要功能: func (bt *Redditbeat) Run(b *beat.Beat) error { ... } 

Stop 函数管理优雅关闭: func (bt *Redditbeat) Stop() { ... } 

Note 1:在Go中没有明确的接口实现。实现了 interface 中的所有方法，即创建一个隐式继承关系. 出于写文档的目的，这是 Beater 接口:type Beater interface {

Run(b *Beat) error

Stop()

}

第 4 段（可获 0.93 积分）

R e; 7个月前 

因此，由于Beat结构实现了Run和Stop，它是一个Beater。Note 2: 在Go中没有类的概念，所以方法不能在一个具体类型上声明。但是，它存在扩展函数的概念：可以添加行为到一个类型（在单个包中）的函数。它需要声明receiver 类型：这是在fun关键字和函数名之间完成的 - 这里是指Redditbeat类型（或者更准确地说，是一个指向Redditbeat类型的指针，但是这里有一个隐式转换）。构造函数和Stop函数可以保持不变，无论什么特性都应该在Run函数中。在这种情况下，功能是调用Reddit REST API并为每个Reddit帖子发送一条消息。

第 5 段（可获 1.59 积分）

R e; 7个月前 

最终代码如下所示：func (bt *Redditbeat) Run(b *beat.Beat) error {

bt.client = b.Publisher.Connect()

ticker := time.NewTicker(bt.config.Period)

reddit := "" + bt.config.Subreddit + "/.json"

client := http.Client {}

for {

select {

case -bt.done:

return nil

case -ticker.C:

}

req, reqErr := http.NewRequest("GET", reddit, nil)

req.Header.Add("User-Agent", "Some existing header to bypass 429 HTTP")

if (reqErr != nil) {

panic(reqErr)

}

resp, getErr := client.Do(req)

if (getErr != nil) {

panic(getErr)

}

body, readErr := ioutil.ReadAll(resp.Body)

defer resp.Body.Close()

if (readErr != nil) {

panic(readErr)

}

trimmedBody := body[len(prefix):len(body) - len(suffix)]

messages := strings.Split(string(trimmedBody), separator)

for i := 0; i  len(messages); i ++ {

event := common.MapStr{

"@timestamp": common.Time(time.Now()),

"type":       b.Name,

"message":    "{" + messages[i] + "}",

}

bt.client.PublishEvent(event)

}

}

}

第 6 段（可获 0.09 积分）

R e; 7个月前 

这里是对最重要的几部分的解释：line 4: 通过连接字符串创建Reddit REST URL，包括配置Subreddit参数。记住，它的默认值已在config.go文件中定义。line 5: 引用httpClient类型line 12: 创建新的HTTP请求。注意Go允许多个返回值。line 13: 如果没有设置标准请求头，Reddit的API将返回429状态码。line 14: Go标准错误不通过异常处理，而是随着常规返回值返回。根据Golang wiki：指示调用者的错误条件，应通过返回错误值来完成line 15: panic() 函数类似于在Java中抛出异常, 被处理时推到栈顶。 有关详细信息，请查看相关文档。line 17: 执行HTTP请求。line 21: 将响应主体读入字节数组。line 22: 关闭主体流。注意defer关键字：defer语句延迟函数的执行，直到环绕的函数返回。line 26: 创建整个响应主体字节数组的切片 - 对数组的一部分的引用。实质上，它删除了前缀和后缀以保持相关的JSON值。之后将字节数组解析成JSON。line 27: 分割切片以单独获取每个JSON片段。line 29: 将消息创建为简单的字典结构。line 34: 发送。

第 7 段（可获 3.11 积分）

R e; 7个月前 

配置, 构建, 运行默认配置参数可以在项目根目录下的redditbeat.yml文件中找到。请注意，redditbeat.full.yml中列出了其他常见的Beat参数，以及相关注释。关于Beats的一个有趣的事情是，他们的消息可以直接发送到Elasticsearch或Logstash进行进一步处理。这在上述配置文件中配置。redditbeat:

period: 10s

output.elasticsearch:

hosts: ["localhost:9200"]

output.logstash:

hosts: ["localhost:5044"]

enabled: true

第 8 段（可获 0.78 积分）

R e; 7个月前 

此配置片段将每10秒循环运行Run方法，并将消息发送到在localhost上运行的Logstash实例在端口5044上。这可以在运行Beat时被覆盖（见下文）。注意：为了使Logstash接受来自Beats的消息，必须安装Logstash Beat插件，并且必须为Beats配置Logstash的input：input {

beats {

port = 5044

}

}

要构建项目，请在项目的根目录中键入make。它将创建一个可以运行的可执行文件。./redditbeat -e -E redditbeat.subreddit=java

-E参数可以覆盖在的redditbeat.yml配置文件中找到的参数（见上文）。在这里，它设置subreddit读为“java”，而不是默认的“elastic”。

第 9 段（可获 1.3 积分）

R e; 7个月前 

输出如下所示:2016/12/17 14:51:19.748329 client.go:184: DBG  Publish: {

"@timestamp": "2016-12-17T14:51:19.748Z",

"beat": {

"hostname": "LSNM33795267A",

"name": "LSNM33795267A",

"version": "6.0.0-alpha1"

},

"message": "{

\"kind\": \"t3\", \"data\": {

\"contest_mode\": false, \"banned_by\": null, 

\"domain\": \"blogs.oracle点抗 \", \"subreddit\": \"java\", \"selftext_html\": null, 

\"selftext\": \"\", \"likes\": null, \"suggested_sort\": null, \"user_reports\": [], 

\"secure_media\": null, \"saved\": false, \"id\": \"5ipzgq\", \"gilded\": 0, 

\"secure_media_embed\": {}, \"clicked\": false, \"report_reasons\": null, 

\"author\": \"pushthestack\", \"media\": null, \"name\": \"t3_5ipzgq\", \"score\": 11, 

\"approved_by\": null, \"over_18\": false, \"removal_reason\": null, \"hidden\": false, 

\"thumbnail\": \"\", \"subreddit_id\": \"t5_2qhd7\", \"edited\": false, 

\"link_flair_css_class\": null, \"author_flair_css_class\": null, \"downs\": 0, 

\"mod_reports\": [], \"archived\": false, \"media_embed\": {}, \"is_self\": false, 

\"hide_score\": false, \"spoiler\": false, 

\"permalink\": \"/r/java/comments/5ipzgq/jdk_9_will_no_longer_bundle_javadb/\", 

\"locked\": false, \"stickied\": false, \"created\": 1481943248.0, 

\"url\": \"\", 

\"author_flair_text\": null, \"quarantine\": false, 

\"title\": \"JDK 9 will no longer bundle JavaDB\", \"created_utc\": 1481914448.0, 

\"link_flair_text\": null, \"distinguished\": null, \"num_comments\": 4, 

\"visited\": false, \"num_reports\": null, \"ups\": 11

}

}",

"type": "redditbeat"

}

组件分享之后端组件——基于Golang实现的高性能和弹性的流处理器benthos

近期正在探索前端、后端、系统端各类常用组件与工具，对其一些常见的组件进行再次整理一下，形成标准化组件专题，后续该专题将包含各类语言中的一些常用组件。欢迎大家进行持续关注。

本节我们分享的是基于Golang实现的高性能和弹性的流处理器 benthos ，它能够以各种代理模式连接各种 源 和 接收器，并对有效负载执行 水合、浓缩、转换和过滤 。

它带有 强大的映射语言 ，易于部署和监控，并且可以作为静态二进制文件、docker 映像或 无服务器函数 放入您的管道，使其成为云原生。

Benthos 是完全声明性的，流管道在单个配置文件中定义，允许您指定连接器和处理阶段列表：

Apache Pulsar, AWS (DynamoDB, Kinesis, S3, SQS, SNS), Azure (Blob storage, Queue storage, Table storage), Cassandra, Elasticsearch, File, GCP (Pub/Sub, Cloud storage), HDFS, HTTP (server and client, including websockets), Kafka, Memcached, MQTT, Nanomsg, NATS, NATS JetStream, NATS Streaming, NSQ, AMQP 0.91 (RabbitMQ), AMQP 1, Redis (streams, list, pubsub, hashes), MongoDB, SQL (MySQL, PostgreSQL, Clickhouse, MSSQL), Stdin/Stdout, TCP UDP, sockets and ZMQ4.

1、docker安装

具体使用方式可以参见该 文档

有关如何配置更高级的流处理概念（例如流连接、扩充工作流等）的指导，请查看 说明书部分。

有关在 Go 中构建您自己的自定义插件的指导，请查看 公共 API。