hello，大家好呀，我是小楼。
上篇文章《一言不合就重构》 说了我最近重构的一个系统，虽然重构完了，但还在灰度，这不，在灰度过程中又发现了一个问题。
这个问题简单说一下背景，如果不明白可以看上篇文章 ，不想看也没关系，这是个通用的解法，后面我会总结抽象下。
在上篇文章的最后提到对每个摘除的地址做决策时，需要顺序执行，且每一个要摘除的地址都要实时获取该集群的地址信息，以便做出是否需要兜底的决策。
当被摘除的机器非常多时，获取地址信息的请求量就会非常大，对注册中心造成了不小的压力。
请求数据源的接口如下所示（其中 cuuid 是集群的 id）
type Read interface {
ListClusterEndpoints(ctx context.Context, cuuid string) ([]ptypes.Endpoint, error)
}
相信大家也能理解这个非常简单的背景并且能想到一些解法。每次决策需要按 cuuid 获取集群，也就是单个单个地获取实时集群地址信息，由于是实时信息，缓存首先排除，其次自然而然地能想到如果能将请求合并一下，是不是就能解决请求量大的问题？
如果只是改逻辑合并一下请求，吭哧吭哧改代码就完了，也不值得写这篇文章了，如何改最少的代码来实现合并请求才是最难的。
那天遇到这个问题，晚上辗转反侧想到了这个解法，其实主要也是参考 Go http client 的实现，都说看源码没用，这不就是用处么？
Read 数据源接口定义保持不变，也就是上层的业务代码完全不用改，只需要把 ListClusterEndpoints 的实现换掉。
我们可以用一个队列把每个请求入队，入队列以后，调用方阻塞，然后起一些协程去队列里取一批请求参数，发起批量请求，响应之后唤醒阻塞的调用方。
为此，我们实现一个可以阻塞并被其他协程唤醒的工具：
type token struct {
value interface{}
err error
}
type Token chan token
func NewToken() Token {
return make(Token, 1)
}
func (t Token) Done(value interface{}, err error) {
t <- token{value: value, err: err}
}
func (t Token) Wait(timeout time.Duration) (value interface{}, err error) {
if timeout <= 0 {
tk := <-t
return tk.value, tk.err
}
select {
case tk := <-t:
return tk.value, tk.err
case <-time.After(timeout):
return nil, ErrTokenTimeout
}
}
其次，定义队列和其他参数：
type DataSource struct {
paramCh chan param
readTimeout time.Duration
concurrency int
step int
}
type param struct {
cuuid string
token Token
}
替换掉原来 ListClusterEndpoints 的实现：
func (p *DataSource) ListClusterEndpoints(ctx context.Context, cuuid string) ([]ptypes.Endpoint, error) {
req := param{
cuuid: cuuid,
token: NewToken(),
}
select {
case p.paramCh <- req:
default:
return nil, fmt.Errorf("list cluster endpoints write channel failed")
}
value, err := req.token.Wait(p.readTimeout)
if err != nil {
return nil, err
}
eps, ok := value.([]ptypes.Endpoint)
if !ok {
return nil, fmt.Errorf("value is not endpoints")
}
return endpoints, nil
}
再起几个协程来处理任务：
func (p *DataSource) startListClusterEndpointsLoop() {
for i := 0; i < p.concurrency; i++ {
go func() {
for {
reqs := p.getListClusterEndpointsReqFromChan()
p.doBatchListClusterEndpoints(reqs)
}
}()
}
}
最关键的是 getListClusterEndpointsReqFromChan 的实现，既不能让协程空跑，这样太消耗cpu，又要能及时地取到一批参数，我们采取的方法是先阻塞地获取一个参数，如果没数据则阻塞，如果有数据，继续取，直到数量达到上限或者取不到数据为止，此时这一批数据就可以批量地进行调用了。
func (p *DataSource) getListClusterEndpointsReqFromChan() []param {
reqs := make([]param, 0)
select {
case req := <-p.paramCh:
reqs = append(reqs, req)
for i := 1; i < p.step; i++ {
select {
case reqNext := <-p.paramCh:
reqs = append(reqs, reqNext)
default:
break
}
}
}
return reqs
}
这个方法很简单，但是有一些要注意的地方，得做好监控，比如调用方单个请求的QPS、RT，实际批量请求的QPS、RT，这样才好计算出处理协程开多少个合适，还有队列写入失败、队列长度等等监控，当容量不足时及时做出调整。
与本文相关的文章也顺便推荐给你，如果觉得还不错，记得关注、点赞、在看、分享。
搜索关注微信公众号"捉虫大师"，后端技术分享，架构设计、性能优化、源码阅读、问题排查、踩坑实践；
在我的Controller中，我通过以下方式在我的index方法中支持HTML和JSON:respond_todo|format|format.htmlformat.json{renderjson:@user}end在浏览器中拉起它时，它会自然地以HTML呈现。但是，当我对/user资源进行内容类型为application/json的curl调用时(因为它是索引方法)，我仍然将HTML作为响应。如何获取JSON作为响应？我还需要说明什么？ 最佳答案 您应该将.json附加到请求的url，提供的格式在routes.rb的路径中定义。这
所以我开始关注ruby，很多东西看起来不错，但我对隐式return语句很反感。我理解默认情况下让所有内容返回self或nil但不是语句的最后一个值。对我来说，它看起来非常脆弱(尤其是)如果你正在使用一个不打算返回某些东西的方法(尤其是一个改变状态/破坏性方法的函数!)，其他人可能最终依赖于一个返回对方法的目的并不重要，并且有很大的改变机会。隐式返回有什么意义？有没有办法让事情变得更简单？总是有返回以防止隐含返回被认为是好的做法吗？我是不是太担心这个了？附言当人们想要从方法中返回特定的东西时，他们是否经常使用隐式返回，这不是让你组中的其他人更容易破坏彼此的代码吗？当然，记录一切并给出
rails中是否有任何规定允许站点的所有AJAXPOST请求在没有authenticity_token的情况下通过？我有一个调用Controller方法的JqueryPOSTajax调用，但我没有在其中放置任何真实性代码，但调用成功。我的ApplicationController确实有'request_forgery_protection'并且我已经改变了config.action_controller.consider_all_requests_local在我的environments/development.rb中为false我还搜索了我的代码以确保我没有重载ajaxSend来发送
给定以下方法:defsome_method:valueend以下语句按我的预期工作:some_method||:other#=>:valuex=some_method||:other#=>:value但是下面语句的行为让我感到困惑:some_method=some_method||:other#=>:other它按预期创建了一个名为some_method的局部变量，随后对some_method的调用返回该局部变量的值。但为什么它分配:other而不是:value呢？我知道这可能不是一件明智的事情，并且可以看出它可能有多么模棱两可，但我认为应该在考虑作业之前评估作业的右侧...我已经在R
我在我的Rails3示例应用程序上使用CarrierWave。我想验证远程位置上传，因此当用户提交无效URL(空白或非图像)时，我不会收到标准错误异常:CarrierWave::DownloadErrorinImageController#createtryingtodownloadafilewhichisnotservedoverHTTP这是我的模型:classPaintingtrue,:length=>{:minimum=>5,:maximum=>100}validates:image,:presence=>trueend这是我的Controller:classPaintingsC
电脑0x0000001A蓝屏错误怎么U盘重装系统教学分享。有用户电脑开机之后遇到了系统蓝屏的情况。系统蓝屏问题很多时候都是系统bug，只有通过重装系统来进行解决。那么蓝屏问题如何通过U盘重装新系统来解决呢？来看看以下的详细操作方法教学吧。 准备工作： 1、U盘一个（尽量使用8G以上的U盘）。 2、一台正常联网可使用的电脑。 3、ghost或ISO系统镜像文件（Win10系统下载_Win10专业版_windows10正式版下载-系统之家）。 4、在本页面下载U盘启动盘制作工具：系统之家U盘启动工具。 U盘启动盘制作步骤： 注意：制作期间，U盘会被格式化，因此U盘中的重要文件请注
我正在研究使用EventMachine支持的twitter-streamrubygem来跟踪和捕获推文。我对整个事件编程有点陌生。我如何判断我在事件循环中所做的任何处理是否导致我落后？有没有简单的检查方法？ 最佳答案 您可以通过使用周期性计时器并打印出耗时来确定延迟。如果您使用的是1秒的计时器，您应该已经过了大约1秒，如果它更长，您就知道您正在减慢react器的速度。@last=Time.now.to_fEM.add_periodic_timer(1)doputs"LATENCY:#{Time.now.to_f-@last}"@
啊，正则表达式有点困惑。我正在尝试删除字符串末尾所有可能的标点符号:ifstr[str.length-1]=='?'||str[str.length-1]=='.'||str[str.length-1]=='!'orstr[str.length-1]==','||str[str.length-1]==';'str.chomp!end我相信有更好的方法来做到这一点。有什么指点吗？ 最佳答案 str.sub!(/[?.!,;]?$/,'')[?.!,;]-字符类。匹配这5个字符中的任何一个(注意，。在字符类中并不特殊)?-前一个字符或组
我是Ruby的新手。我试过查看在线文档，但没有找到任何有效的方法。我想在以下HTTP请求botget_response()和get()中包含一个用户代理。有人可以指出我正确的方向吗？#PreliminarycheckthatProggitisupcheck=Net::HTTP.get_response(URI.parse(proggit_url))ifcheck.code!="200"puts"ErrorcontactingProggit"returnend#Attempttogetthejsonresponse=Net::HTTP.get(URI.parse(proggit_url)
在我的路线文件中我有:match'graphs/(:id(/:action))'=>'graphs#(:action)'如果是GET请求(工作)或POST请求(不工作)，我想匹配它我知道我可以使用以下方法在资源中声明POST请求:post'/'=>:show,:on=>:member但是我怎样才能为比赛做到这一点呢？谢谢。 最佳答案 如果你同时想要POST和GETmatch'graphs/(:id(/:action))'=>'graphs#(:action)',:via=>[:get,:post]编辑默认值可以设置如下match'g