假设X是一个离散型随机变量，其取值集合为$\varkappa$

信息量

针对事件$x_0$，它发生的“惊讶程度”或不确定性，也就是信息量，通过下面的公式计算。

$I(x_0) = -log_2 \, p(x_0)$

熵（信息熵）

对于一个随机变量X而言，它的所有可能取值的信息量的期望$E[I(x)]$就称为熵

离散变量

$H(X)= E_p log_2 \frac {1}{p(x)} = -\sum_{x\in \varkappa} p(x) log_2 \, p(x)$

连续变量

$H(X) = - \int_{x \in \varkappa} \, p(x) \, log_2 p(x) \, dx$

相对熵(Relative Entropy)

相对熵(Eelative Entropy)又称为KL散度（Kullback-Leibler divergence），KL距离，是两个随机分布间距离的度量。记作$D_{KL}(p||q)$

$\begin{aligned} D_{KL}(p||q) &= E_p[log \, \frac {p(x)} {q(x)}] \\ &= \sum_{x \in \varkappa} \, p(x) \, log \, \frac {p(x)} {q(x)} \\ &= \sum_{x \in \varkappa} [p(x)\,log\,p(x) - p(x)\,log\,q(x)] \\ &= \sum_{x \in \varkappa} p(x)\,log\,p(x) - \sum_{x \in \varkappa} p(x)\,log\,q(x) \\ &= -H(p) - \sum_{x \in \varkappa} p(x)\,log\,q(x) \\ &= -H(p) + E_p [-log\,q(x)] \\ &= H_p(q) - H(p) \end{aligned}$

上述公式中log = log_2，但是在代码实现中都以$e$为底数

假设$p$为真实概率分布，$q$为我们假设的概率分布

• 当$p=q$时，显然$D_KL(p||q)$=0
• $H(p)$ 表示对真实分布p所需要的最小编码bit数
• $H_q(p)$ 表示在$p$分布下，使用$q$进行编码所需要的bit数量
• $D_KL(p||q)$表示在真实分布$p$的前提下，使用$q$进行编码相对于$p$进行编码(最优编码)多出来的bit数

交叉熵(Cross Entropy)

$CEH(p, q) = E_p[-log\,q] = - \sum_{x \in \varkappa} p(x) log\,q(x) = H(p) + D_{KL}(p||q)$

在$p$为真实概率分布的前提下，$H(p)$可以看作常数，此时交叉熵和相对熵在行为上表现一致，都反映分布$p$和$q$之前的相似程度。所以一般在机器学习中，都直接优化交叉熵。

逻辑回归(Logistic Regression)

• p: 真实样本分布，服从参数为p的0-1分布
• q: 待估计的模型，服从参数为q的0-1分布

定义假设函数（hypothesis function）为

$h_{\theta} = \frac {1} {1+ e ^{ -{\theta}^T x }}$

逻辑回归本质就是2分类问题

$P(\hat{y}| x^{(i)};\theta) = \begin{cases} h_{\theta}(x^{(i)}) &\text{if } \hat{y} = 1 \\ 1- h_{\theta}(x^{(i)}) &\text{if } \hat{y} = 0 \end{cases}$

上述公式可以写为更一般的形式

$P(\hat{y}| x^{(i)};\theta) = \hat{y} \, h_{\theta}(x^{(i)}) + (1- \hat{y}) (1- h_{\theta}(x^{(i)}))$

带入至交叉熵公式中

$\begin{aligned} Loss(\hat{y}, y) \\ &= CEH(p, q) \\ &= - p(\hat{y})\,log\,q(y) \\ &= - [P_p(\hat{y}=1)]\,logP_q(\hat{y}=1) + P_p(\hat{y}=0)\,log\,P_q(\hat{y}=0)] \\ &= -[\hat{y}\,log\,h_{\theta}(x)+(1-\hat{y})\,log\,(1-h_{\theta}(x))] \end{aligned}$

对于m个样本取均值

$\frac {1}{m} \sum_{i=1}^{m}[y^{(i)} h_{\theta}(x)^{(i)} + (1-y^{(i)}) (1-h_{\theta}(x)^{(i)})]$

验证

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

import tensorflow as tf
import numpy as np
lables = np.array([0,1,1,1,0], dtype=np.float32)
predictions = np.array([-5,6,5,2,-1], dtype=np.float32)

def sigmoid(x):
    return 1 / (1 + np.exp(-x))
    
def sigmoid_corss_entropy(lables, predictions):
    logits = sigmoid(predictions)
    ces =  - lables * np.log(logits) - (1 - lables) * np.log(1-logits)
    return ces

np_sigmoid_corss_entropy = sigmoid_corss_entropy(lables, predictions).mean()

with tf.Session() as sess:
    tf_sigmoid_cross_entropy_tensor = tf.losses.sigmoid_cross_entropy(multi_class_labels=tf.constant(lables), logits=tf.constant(predictions))
    tf_sigmoid_cross_entropy = sess.run(tf_sigmoid_cross_entropy_tensor)

print(np_sigmoid_corss_entropy) # 0.0912192
print(tf_sigmoid_cross_entropy) # 0.0912192

人的思维是具有连续性的，但是普通的神经网络并不不能维持记忆。循环神经网络的主要用途是处理和预测 序列数据。

序列是一个比较抽象的概念，比如

• 股票价格曲线对应的时间轴
• 正弦曲线的x轴
• 一张图片的以宽度作为滑动窗口的序列
• 一段文字
• 一段语音

处理：即将指定维度的特征经过神经网络处理之后，再处理为原来的维度
预测：使用神经网络的输出直接做预测

RNN

RNNs看成是一个普通的网络做了多次复制之后叠加在一起

• RNN内部状态通过向量表示，维度h
• 输入向量维度 x
• 输入：上一个时刻的状态+当前时刻的输入，h+x
• 输出：输出为当前时刻的状态，维度h。即做一次 tanh(h+x->x 的全连接)

还是很抽象，看一下前向传播的具体过程普通

RNN出现的主要目的是将以前的信息联系到现在，从而解决现在的问题。

但是随着预测信息和相关信息间的间隔增大， RNNs 很难去把它们关联起来了。

LSTM 长短期记忆网络（Long Short Term Memory networks）

遗忘门

传入门

更新门

其他变种

1. 双向神经网络 bidirectional RNN，两个单向神经网络的结合，比如 完形填空，需要预测的值和前后都相关
2. 深层神经网络，增强模型表达能力，在每一个时刻上将循环体重复多次，内部结构维度一样，但是值不一样

Refercnce

（译）理解 LSTM 网络 （Understanding LSTM Networks by colah）

Signal

软中断信号（signal，又简称为信号）用来通知进程发生了异步事件。进程之间可以互相通过系统调用kill发送软中断信号。内核也可以因为内部事件而给进程发送信号，通知进程发生了某个事件。注意，信号只是用来通知某进程发生了什么事件，并不给该进程传递任何数据。

在UnixSignal中提到了很多Signal，这里主要关注以下两个:

• SIGTERM 用来通知进程停止。Kill 命令默认就是发送该信号。这是一种相对礼貌的方式，进程可以对改信号做出处理，比如释放正在使用的资源之后再退出，或者直接忽略。
• SIGKILL 用来让进程立即停止。Kill -s SIGKILL $pid。和SIGTERM不同，该信号不能被进程监听或者忽略。

Graceful Shutdown

生产环境中我们的应用一般通过守护进程来托管。下面看下不同的守护进程是否如何使用上面的信号的

• supervisord 停止服务时，默认使用SIGTERM，并且等待10s，如果进程还在运行，则使用SIGKILL。参考stopsignal和stopwaitsecs配置
• docker 容器中的主进程会接受的SIGTERM，默认10s之后，发送SIGKILL
• pm2 发送SIGINT给相关子进程，没有和上面两个做类似的动作，不过官方文档中提到了应用如何自己实现优雅关闭

上面提到的都是守护进程方面如何优雅的停止进程。但是仍然有一个问题，以web server为例，在SIGTERM超时到SIGKILL过程中，还是会有请求不断的过来。那么在集群环境下，如何做到重启一个节点对生产无感知?

以zk作为服务发现为例，可以用以下步骤:

1. zk通知consumer节点，有一个server即将关闭。即将关闭的消息需要同步给很多consumer，如果确保所有consumer节点均收到消息之后修改本地负载均衡池的话再执行server关闭的话，保证一致性带来的成本会比较高，可以简单设置一个超时时间。
2. 在同步下线消息之后，consumer应该确保不再向即将下线的server发起新的请求。
3. 一般server端都会有服务端超时时间，即处理一个请求，超过一定时间，即抛出Timeout异常。上文中提到的SIGTERM到SIGKILL超时10s对于一个请求来说已经绰绰有余。server接受到关闭的信号之后，尽可能处理好进程中正在处理的请求，到了超时时间之后再执行关闭动作

Referecne

• Unix signal
• Linux 信号signal处理机制
• Graceful shutdown in node.js

前言

对象池作为一种设计模式，很多语言都有对应的实现。比较著名的应该就是commons-pool。实际开发中，可能接触的比较的多的就是db的连接池。以下摘自wiki：

The object pool pattern is a software creational design pattern that uses a set of initialized objects kept ready to use – a “pool” – rather than allocating and destroying them on demand. A client of the pool will request an object from the pool and perform operations on the returned object. When the client has finished, it returns the object to the pool rather than destroying it; this can be done manually or automatically.
Object pools are primarily used for performance: in some circumstances, object pools significantly improve performance. Object pools complicate object lifetime, as objects obtained from and returned to a pool are not actually created or destroyed at this time, and thus require care in implementation.

通过node-pool简单了解下object-pool的具体实现

Project Struct

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

├── CHANGELOG.md
├── Makefile
├── README.md
├── index.js
├── lib
│   ├── DefaultEvictor.js
│   ├── Deferred.js
│   ├── Deque.js
│   ├── DequeIterator.js
│   ├── DoublyLinkedList.js
│   ├── DoublyLinkedListIterator.js
│   ├── Pool.js
│   ├── PoolDefaults.js
│   ├── PoolOptions.js
│   ├── PooledResource.js
│   ├── PooledResourceStateEnum.js
│   ├── PriorityQueue.js
│   ├── Queue.js
│   ├── ResourceLoan.js
│   ├── ResourceRequest.js
│   ├── errors.js
│   ├── factoryValidator.js
│   └── utils.js
├── package.json
└── test
    ├── doubly-linked-list-iterator-test.js
    ├── doubly-linked-list-test.js
    ├── generic-pool-acquiretimeout-test.js
    ├── generic-pool-test.js
    ├── resource-request-test.js
    └── utils.js

整个结构可以划分如下：

• 基本数据结构
  • DoublyLinkedList.js 双链表
  • DoublyLinkedListIterator.js
  • Deque.js 双端对列
  • DequeIterator.js
  • Queue.js
  • PriorityQueue.js
• Promise相关的封装
  • Deferred.js
• Pool相关
  • PooledResource.js
  • PooledResourceStateEnum.js
  • ResourceRequest.js
  • ResourceLoan.js
  • PoolDefaults.js
  • PoolOptions.js
  • DefaultEvictor.js 默认的evictor policy实现
  • Pool.js pool的主要逻辑实现

pool.js

attribute

• waitingClientsQueue:ProtorityQueue 请求资源中的clients
• factoryCreateOperations:Set 正在处理的创建资源的请求
• factoryDestroyOperations:Set 正在处理的销毁资源的请求
• avaiableObjects:Deque 状态为idel的资源，因为存在fifo的配置选项，所以存储时选择的数据结构为双端对列
• testOnBorrowResources:Set 在获取之前处于testing中的资源
• testOnReturnResources:Set 在返回pool之前的testing资源
• validationOperations: Set
• allObjects:Set 当前pool中所有的没有被destroy的资源
• resourceLoans：Map，结构为

xxxOperations:Set的作用为暂存执行create、desitory、validation中的请求

getter

• potentiallyAllocableResourceCount： 可以被分配的资源的数量，为以下之和
  • availableObjects
  • testOnBorrowResources
  • testOnReturnResources
  • factoryCreateOperations
• count = allObjects + factoryCreateOperations：当前已经创建的资源和即将创建的资源之和
• spareResourceCapacity = config.max - count：当前还可以创建的数量

function

dispatchPooledResourceToNextWaitingClient(pooledResource)

1. 从watingClientQueue dequeue一个request，如果request不存在，则将resource标记为idle并添加至availableObjects中，返回fasle
2. 如果存在request，构造ResourceLoan，并添加loan对象至resourceLoans中
3. pooledResouce标记为allocated
4. 执行request.resolve
5. 返回true

dispatchResource(pooledResouce)

1. 从availableObjects中取出resource，并执行dispatchPooledResourceToNextWaitingClient

dispense()

1. 记录waitingClientsQueue.length为局部变numWaitingClients，如果numWaitingClients<1则直接return
2. 记录numWaitingClients-potentiallyAllocableResourceCount的差值为resourceShotfall，即对于watingClient实际缺少的resource的数量
3. resourceShotfall和spareResourceCapacity中取最小值，即本次执行函数时，可以创建resource的数量
4. for循环执行createResource
5. testOnBorrow设置为true，计算出实际多少resource需要转入test状态，for循环执行testOnBorrow
6. testOnBorrow设置为false时，取availableObjects和numWaitingClients中较小的数值，for循环执行dispatchResource

createResource()

1. 执行factory.create，resolve之后
   1. 构造PooledResource实例
   2. allObjects.add
   3. dispatchPooledResourceToNextWaitingClient
2. 如果reject，则执行dispense，实现递归

testOnBorrow()

1. availableObjects中取出resource并标记状态为test
2. 添加址testOnBorrowResources中
3. 执行factory.validate，promise reoslve的结果为boolean类型，resolve结束时，从testOnBorrowResources移除
   1. 结果为true，执行dispatchPooledResourceToNextWaitingClient
   2. 结果为false，标记resource为invalidate；destroy resource；dispense，实现递归

acqurie(priority)

1. 判断是否draining，如果是则reject error
2. 判断waitingClientsQueue是否已经大于maxWaitingClients，如果是，则reject error
3. 构造ResourceRequest，并enqueue至waitingClientsQueue中
4. 执行dispense
5. 返回resourceRequest.promise

release(resource)

1. 在createResource的逻辑中，会将分配出去的resource添加至resourceLoans中，release的主要作用也是从resourceLoans将resouce移除
2. 标记resource状态为idel，并归还至availableObjects

总结

1. 整个项目3.x版本重构过了，使用了es6的class来构建模块，可以通过类图理解结构，好评
2. factory接口中定义的create、destory、validate的均为异步函数，关于在判断数量时均需要考虑正在create、destory和validte的数量。这里是factory中的promise通过set保存
3. resource对象有很多状态，对应的pool中就要有相应的容器来存放相应状态的resource或者promise，理解pool的关键就在此。
4. 感觉好难用图来表述逻辑，只能用文字了。

Reference

• Generic Pool
• commons-pool

Key Concepts

• buckets 每个stat都有各自的bucket，无需事先定义
• values 每个stat都会有一个value

Line Protocol

1

<metricname>:<value>|<type>

1

echo "foo:1|c" | nc -u -w0 127.0.0.1 8125

Workflow

Project Struct

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

├── CONTRIBUTING.md
├── Changelog.md
├── Dockerfile
├── LICENSE
├── README.md
├── backends	在得到聚合的metric后，针对不同的后端各自的逻辑
├── bin
├── debian
├── docker-compose.yml
├── docs
├── exampleConfig.js
├── exampleProxyConfig.js
├── examples
├── lib
│   ├── config.js	加载config处理的逻辑
│   ├── helpers.js	help function
│   ├── logger.js	自定义的logger类
│   ├── mgmt_console.js	mangenment consoel的逻辑函数
│   ├── mgmt_server.js	mangement server启动函数
│   ├── process_metrics.js		加工处理metrics
│   ├── process_mgmt.js	对主进程的一些基本设置
│   └── set.js		自己实现的set结构
├── package.json
├── packager
├── proxy.js
├── run_tests.sh
├── servers	metric server，包含tpc和udp两种
├── stats.js	入口文件，包含大部分启动逻辑
├── test
└── utils

stats.js

应用启动的入口文件，主要:

1. 声明一些全局存储变量
2. 声明一些根据config启动server的函数
3. 根据process.args[2]即配置文件路径启动应用

具体的代码的逻辑大致如下:

1. config.configFile(configPath:String)，加载配置
2. process_mgmt.init(config:Object)，设置进程相关的一些基本参数
3. 根据config.prefixStats(String:’statsd’)来设置一些内置metric的前缀
4. 初始化counter内bad_lines_seen、packets_received、metrics_received为0
5. 通过config.keyNameSanitize判断是否需要转义metric key
6. 声明handlePacket(msg:Buffer,rinfo:Object)，主要作用是接受报文，根据line protocol解析，然后将结果写入一开始申明的全局变量
7. 根据config.servers的配置，从servers目录中加载并启动server，server监听的回调函数就是上一步中声明的handlePacket
8. mgmt_server.start(conifg:Object,on_data_callback:Functioin,on_error_callback)，启动一个tcp的mangement的server
9. 设置percentThreshold，默认[90]
10. 设置flushInterval，默认10s
11. 根据config.backends设置后端的server，默认为backends/graphite。注意：如果我们需要测试，可以设置为backends/console
12. 通过在flushMetrics函数中使用递归调用setTimeout实现flush timer。flushMetrics的大致逻辑内部逻辑为：
    1. 先构造metrics_hash对象
    2. 在backendEvents:EventEmiter中注册flush事件监听，如果config设置了相关的delete配置，则删除相关key，否则置为0或者空数组
    3. 通过process_metrics对构造的metrics_hash对象根据配置做一些加工计算处理，结束之后触发的backendEvents的flush事件
    4. 再次注册setTimeout，进行下一次flush

lib/process_metrics.js

主要作用是对stats.js中flushMetrics函数中传入的metric_hash做计算加工，并将结果返回出去

具体代码逻辑如下：

1. 声明基本的局部存储变量
2. 遍历counter，将将每秒的结果计算至counter_rates中
3. 遍历timers，每个timer对应的是数组，形如bar: [200, 198, 199]这样。因为timer表示的内容比较丰富，所以计算会多一些，timers中的计算结果最终会计算至timer_data中。比如timers: { bar: [ 100, 200 ] }，至少会计算出

1
2
3
4
5
6
7
8
9

std: 50,
upper: 200,
lower: 100,
count: 2,
count_ps: 0.2,
sum: 300,
sum_squares: 50000,
mean: 150,
median: 150

然后根据pctThreshold参数，计算出各个不同分位的指标，像这样

1
2
3
4
5

count_$pct
mean_$pct
upper_|lower_$pct
sum_$pct
sum_squares_$pct

值得注意的是，关于$pct的相关的指标计算是将排序后，计算落在$pct内的点的计算

做个简单的测试

1

echo "foo:1|c\nbar:1|ms\nbar:2|ms\nbar:3|ms\nbar:4|ms\nbar:5|ms\n" | nc -u -w0 127.0.0.1 8125

console的backend显示如下:

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

{ counters:
   { 'statsd.bad_lines_seen': 0,
     'statsd.packets_received': 1,
     'statsd.metrics_received': 6,
     foo: 1 },
  timers: { bar: [ 1, 2, 3, 4, 5 ] },
  gauges: {},
  timer_data:
   { bar:
      { count_45: 2,
        mean_45: 1.5,
        upper_45: 2,
        sum_45: 3,
        sum_squares_45: 5,
        std: 1.4142135623730951,
        upper: 5,
        lower: 1,
        count: 5,
        count_ps: 0.5,
        sum: 15,
        sum_squares: 55,
        mean: 3,
        median: 3 } },
  counter_rates:
   { 'statsd.bad_lines_seen': 0,
     'statsd.packets_received': 0.1,
     'statsd.metrics_received': 0.6,
     foo: 0.1 },
  sets: {},
  pctThreshold: [ 45 ] }

backends/console.js

Flush stats to graphite

每个backend中的代码均需要实现init方法，在stats.js中通过loadBackend函数调用。init方法中传入的参数有：

1. startup_time, 启动时间
2. config, 
3. backendEvents, 
4. l，日志对象

init函数按照约定都需要同步返回true

比如backends/console.js，init中初始化了ConsoleBackend的一个实例，在构造函数中注册了对参数backendEvents的flush和flush事件

backends/graphite.js

Metric

对发送至graphite的指标数据做了封装

1
2
3

constructor(key:String,val:Number,ts:Timestamp)
toPickle():String	在graphite使用pickle protocol时使用
toText():String

Stats

指标数据的集合，提供了add方法和toText和toPickle这两个序列化实现

init

1. 设置默认host、port、protocol
2. 设置不同指标的默认的前缀和后缀

flush_stats

1. 分别遍历metrics数据中的counters、timer_data、gauges、sets，统一添加至Stats的类实例中，方便后续统一做序列化处理
2. 在添加时stats实例中，会根据统一的前缀和后缀和命名空间做一些指标名称的处理

post_stats

1. 根据config中的host、port建立socket连接
2. 连接成功时，在stats实例中添加统计信息，根据protocol选择序列化方式并写入socket
3. 写入成功之后更新graphiteStats的统计信息，失败也是一样

总结：

1. 代码很简单，由于是一个定时的flush的机制，将很多状态直接存储在对象中
2. backend/*的解耦了process metrics和flush backend，backend/console在开发时用起来很方便
3. 作为一个daemon process，通过mangent server提供简单tcp接口来反馈当前的统计信息。
4. 指标的量很大，所以在协议设计方面尽可能简单，并且对于不论是handlePacket和flush backend时，都尽可能使用了batch
5. 在metrics的架构中，通常作为缓冲层存在。将大量的point的请求在时间维度做聚合，也是batch思想的体现

nodejs的应用一般都用pm2托管，但是pm2本身的日志处理比较弱，时间久了日志文件会变得很大，需要一些日志切割的策略。

linux一般使用logrotate来作日志切割，相关介绍见understanding-logrotate-utility

demo配置如下，假设你的nodejs应用使用www-data启动：

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

${loggerRoot}/*.log {
    rotate 7
    daily
    dateext
    dateformat .%Y%m%d
    compress
    missingok
    notifempty
    sharedscripts
    postrotate
        su -l www-data -c 'pm2 reloadLogs'
    endscript

    su www-data www-data
};

原理就是在logrotate之后调用pm2 reloadLogs

Anatomy of a script

当你创建了你的hubot之后，生成器同样创建了一个scripts目录。你如果打开看看，会发现一样示例脚本。为了让脚本生效，你需要：

• 脚本需要位于hubot的脚本加载目录中(默认为src/scripts和scripts)
• .coffee或.js文件
• export为一个函数

export为一个函数，即：

1
2

module.exports = (robot) ->
  # your code here

参数robot是的一个robot的一个实例。现在我们可以干一些碉堡了的事情了。

Hearing and responding

既然这是一个聊天机器人，那么最常见的互动方式就是基于消息的。Hubot可以hear房间中说的消息，也可以直接respond。这些方法都接受一个正则表达式和一个回调函数作为参数。例如：

1
2
3
4
5
6

module.exports = (robot) ->
  robot.hear /badger/i, (res) ->
    # your code here

  robot.respond /open the pod bay doors/i, (res) ->
    # your code here

robot.hear /badger/，回调函数会在任何任何消息文本匹配时之行。比如：

• Stop badgering the witness
• badger me
• what exactly is a badger anyways

robot.respond /open the pod bay doors/i的回调函数仅在机器人的名字或别名在消息文本前面的时候执行。如果机器人的名字时HAL，别名时/，这些情况下回调触发：

• hal open the pod bay doors
• HAL: open the pod bay doors
• @HAL open the pod bay doors
• /open the pod bay doors

这些情况下不会触发：

• HAL: please open the pod bay doors
  • 因为respond需要文本信息之前跟着机器人名称
• has anyone ever mentioned how lovely you are when you open the pod bay doors?
  • 缺少机器人名称

Send & reply

res参数是Response的一个实例(historically, this parameter was msg and you may see other scripts use it this way)。你可以通过它send消息回去，emote消息（如果你的adapter支持的话），或者reply那个发送消息的用户。例如：

1
2
3
4
5
6
7
8
9

module.exports = (robot) ->
  robot.hear /badger/i, (res) ->
    res.send "Badgers? BADGERS? WE DON'T NEED NO STINKIN BADGERS"

  robot.respond /open the pod bay doors/i, (res) ->
    res.reply "I'm afraid I can't let you do that."

  robot.hear /I like pie/i, (res) ->
    res.emote "makes a freshly baked pie"

robot.hear /badgers/的回调不管谁发送的消息直接回复，”Badgers? BADGERS? WE DON’T NEED NO STINKIN BADGERS”。

如果一个用户Dave说 “HAL: open the pod bay doors”, robot.respond /open the pod bay doors/i的回调函数就会发送消息”Dave: I’m afraid I can’t let you do that.”

Capturing data

至今，我们的脚本都是静态的回复，相对比较无趣一点。res.match包含了消息中正则匹配的部分。这是JavaScript的特性，返回一个数组，索引为0的是匹配正则表达式的全文本。比如：

1
2

robot.respond /open the (.*) doors/i, (res) ->
  # your code here

如果Dave说”HAL: open the pod bay doors”, res.match[0]是”open the pod bay doors”, res.match[1]就是”pod bay”。现在，我们可以做一些更动态的事情了：

1
2
3
4
5
6

robot.respond /open the (.*) doors/i, (res) ->
  doorType = res.match[1]
  if doorType is "pod bay"
    res.reply "I'm afraid I can't let you do that."
  else
    res.reply "Opening #{doorType} doors"

Making HTTP calls Unmaintained

JSON

XML

Screen scraping

Random

一个常见的场景就是听到或者相应一个命令，从数组中随即发送图片或者文本。JavaScript和CoffeeScript并没有提供什么方法，所以Hubot提供了一个方便的方法:

1
2
3

lulz = ['lol', 'rofl', 'lmao']

res.send res.random lulz

Topic

如果adapter支持的话，Hubot可以对房间的主题变更作出相应的反应。

1
2
3

module.exports = (robot) ->
  robot.topic (res) ->
    res.send "#{res.message.text}? That's a Paddlin'"

Entering and leaving

如果adapter支持，Hubot可以看到用户进入和离开。

1
2
3
4
5
6
7
8

enterReplies = ['Hi', 'Target Acquired', 'Firing', 'Hello friend.', 'Gotcha', 'I see you']
leaveReplies = ['Are you still there?', 'Target lost', 'Searching']

module.exports = (robot) ->
  robot.enter (res) ->
    res.send res.random enterReplies
  robot.leave (res) ->
    res.send res.random leaveReplies

Custom Listeners

上面涵盖了普通用户的大部分功能需求 (hear, respond, enter, leave, topic)，有时候，我们需要特别的匹配逻辑。如果这样，你可以使用listen来制定自定义的函数，而不是正则表达式。

如果想回调函数执行，match函数必须返回可以转化为true的值，返回值会传递给response.match

1
2
3
4
5
6
7
8
9

module.exports = (robot) ->
  robot.listen(
    (message) -> # Match function
      # Occassionally respond to things that Steve says
      message.user.name is "Steve" and Math.random() > 0.8
    (response) -> # Standard listener callback
      # Let Steve know how happy you are that he exists
      response.reply "HI STEVE! YOU'RE MY BEST FRIEND! (but only like #{response.match * 100}% of the time)"
  )

Events

Hubot还可以对事件作出相应，这可以用来script之间传递数据。robot.emit和robot.on通过Nodej.js的EventEmitter封装。

一个例子就是有一个脚本和服务互动，当请求来的时候触发事件。比如，我们可以有一个脚本从Github的post-commit钩子接受数据，触发commit事件，另一个脚本处理这些commits。

1
2
3
4
5
6
7
8

# src/scripts/github-commits.coffee
module.exports = (robot) ->
  robot.router.post "/hubot/gh-commits", (req, res) ->
    robot.emit "commit", {
        user    : {}, #hubot user object
        repo    : 'https://github.com/github/hubot',
        hash  : '2e1951c089bd865839328592ff673d2f08153643'
    }

1
2
3
4
5

# src/scripts/heroku.coffee
module.exports = (robot) ->
  robot.on "commit", (commit) ->
    robot.send commit.user, "Will now deploy #{commit.hash} from #{commit.repo}!"
    #deploy code goes here

如果提供事件，非常建议包含一个在数据中包含一个hubot用户或者房间，这样允许hubot来在聊天中通知用户或房间。

Error Handling

没有代码是完美的，错误和异常都是可以接受的。先前，未捕获的异常会导致hubot实例crash。Hubot现在包含了uncaughtException处理，他来提供脚本来做一些异常处理。

1
2
3
4
5
6
7

# src/scripts/does-not-compute.coffee
module.exports = (robot) ->
  robot.error (err, res) ->
    robot.logger.error "DOES NOT COMPUTE"

    if res?
      res.reply "DOES NOT COMPUTE"

这里你可以做任何处理处理，但是如果你想做一些额外的补救措施、记录日志的话，最好是异步的代码。否则，你可能会遇到递归错误而且还不知道。

会有一个error事件触发，用错误处理函数来消费这个事件。因此，不管会不会发生，你都要处理好自己的异常，并且自己触发他们。第一个参数是触发的错误对象，第二个是可选参数来描述错误。用上面一个例子：

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

robot.router.post '/hubot/chatsecrets/:room', (req, res) ->
  room = req.params.room
  data = null
  try
    data = JSON.parse req.body.payload
  catch err
    robot.emit 'error', err

  # rest of the code here


robot.hear /midnight train/i, (res)
  robot.http("https://midnight-train")
    .get() (err, res, body) ->
      if err
        res.reply "Had problems taking the midnight train"
        robot.emit 'error', err, res
        return
      # rest of code here

第二个例子中，很值得思考下用户会看到什么样子的信息。如果你有一个错误处理函数来回复用户，你可能不需要添加一个自定义的错误提示给用户，但是这个也取决于你对异常提示有多公开。

Persistence

Hubot有一个基于内存的key-value存储，通过robot.brain来存储、设置数据。

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

robot.respond /have a soda/i, (res) ->
  # Get number of sodas had (coerced to a number).
  sodasHad = robot.brain.get('totalSodas') * 1 or 0

  if sodasHad > 4
    res.reply "I'm too fizzy.."

  else
    res.reply 'Sure!'

    robot.brain.set 'totalSodas', sodasHad+1
robot.respond /sleep it off/i, (res) ->
  robot.brain.set 'totalSodas', 0
  msg.reply 'zzzzz'

如果脚本需要寻找用户信息，有一些方法在robot.brain中可以用来通过id、name或者模糊匹配来查找一个或多个用户:userForName, userForId, userForFuzzyName, usersForFuzzyName。

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

module.exports = (robot) ->

  robot.respond /who is @?([\w .\-]+)\?*$/i, (res) ->
    name = res.match[1].trim()

    users = robot.brain.usersForFuzzyName(name)
    if users.length is 1
      user = users[0]
      # Do something interesting here..

      res.send "#{name} is user - #{user}"

LISTENER METADATA

出了正则表达式和回调函数，hear和respond接受一个可选的Object参数，可以很容易的对即将创建的Listener对象添加metadata信息。metadata可以很容易的扩展脚本信息而不需要修改脚本本身。

最重要而且最常见的metadata的key是id，每个Listener都应有被给予一个唯一的id(option.id，默认为null)。通过模块名来划分命名空间(‘my-module.my-listener’)。这些名称允许其他脚本很轻松的定位listeners，扩展像authorization和rate limiting的功能也不需要额外的函数。

额外的扩展可能需要定义额外的metadata key。

提前看个例子：

1
2
3
4
5
6

module.exports = (robot) ->
  robot.respond /annoy me/, id:'annoyance.start', (msg)
    # code to annoy someone

  robot.respond /unannoy me/, id:'annoyance.stop', (msg)
    # code to stop annoying someone

这些定义允许你扩展一些新的行为，比如：

• authorization策略，允许annoyers组的每个人执行annoyers.*命令
• rate limiting:30分钟内只能调用annoyance.start一次

MIDDLEWARE

有三种类型的中间件: Receive, Listener and Response.

Receive 中间件在 listeners 检查之前运行
Listener 中间件运行在每个listener匹配消息之后
Response 中间件在每个消息发送出去时运行

Execution Process and API

和Express middleware相似,
Hubot按定义顺序执行中间件。每个中间件通过next继续中间件，通过done打断中间件。

Middleware调用的时候有以下参数：

• context
  • 详见每个不同中间件的的API
• next
  • 一个没有任何额外属性的函数，用来被执行继续下一个中间件或者执行Listener的回调函数。
  • next调用时有一个可选参数，可以是done函数，或者是一个新的最终会调用done的函数。如果参数未指定，则认为传入了done
• done
  • 一个没有任何额外属性的函数。当需要结束中间件调用和结束整个函数时调用。
  • 调用时没有参数

每个中间键都接受相同的API签名，context、next和done。不同类型的中间件在context中接受不同的信息。

Error Handling

对于同步的中间件（不会产生事件循环），hubot会自动捕获错误并且触发error事件。Hubot还会调用最近的done回调来结束中间件调用。异步中间件应该自己捕获异常，触发error事件，调用done函数。任何未捕获的异常都会打断中间件的所有回调。

LISTENER MIDDLEWARE

Listener中间件在匹配消息和执行listener之间插入逻辑。这允许你为每个匹配的脚本创建扩展。比如，集中的认证策略、调用限制、日志、指标。Middleware的实现和其他脚本一样，但是并不是使用hear和respond，中间件使用listenerMiddleware

Listener Middleware Examples

完整的例子可以参见hubot-rate-limit。

一个记录执行命令的中间件：

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

module.exports = (robot) ->
  robot.listenerMiddleware (context, next, done) ->
    # Log commands
    robot.logger.info "#{context.response.message.user.name} asked me to #{context.response.message.text}"
    # Continue executing middleware
    next()
```    
   
这个例子中，每个匹配的消息都会写下日志信息。

更加复杂的调用限制的例子：

```js
module.exports = (robot) ->
  # Map of listener ID to last time it was executed
  lastExecutedTime = {}

  robot.listenerMiddleware (context, next, done) ->
    try
      # Default to 1s unless listener provides a different minimum period
      minPeriodMs = context.listener.options?.rateLimits?.minPeriodMs? or 1000

      # See if command has been executed recently
      if lastExecutedTime.hasOwnProperty(context.listener.options.id) and
         lastExecutedTime[context.listener.options.id] > Date.now() - minPeriodMs
        # Command is being executed too quickly!
        done()
      else
        next ->
          lastExecutedTime[context.listener.options.id] = Date.now()
          done()
    catch err
      robot.emit('error', err, context.response)

这个例子中，中间件检查listener是否在最近的1s中被调用。如果有，中间件里面调用done，阻止listenr的回调调用。如果listenr允许执行，中间件在next中添加done，这样就能调用记录结束时间。

这个例子同样展示了通过特定的metadata可以创建出很有用的扩展：使用限制中间件，脚本开发人员可以通过设置listener option参数创建调用限制的命令。

1
2
3
4

module.exports = (robot) ->
  robot.hear /hello/, id: 'my-hello', rateLimits: {minPeriodMs: 10000}, (msg) ->
    # This will execute no faster than once every ten seconds
    msg.reply 'Why, hello there!'

Listener Middleware API

Listener中间件的回调函数接受3个参数:context、next和done。

context包含这些字段：

• listener
  • options: 一个Object对象，包含listener中定义的metadata。
• response
  • 所有的response API都包含
  • 中间件用一些额外的信息来装饰（不是修改）response（比如为response.message.user添加LDAP groups信息）
  • 注意：文本信息(response.message.text)应该被考虑为不要改变

RECEIVE MIDDLEWARE

Receive中间件在所有的listeners执行之前。很适合用来实现黑名单功能而不用考虑ID，metrics等。

Receive Middleware Example

示例中间件禁止特定的用户使用包括hear在内的功能。如果一个用户想使用一个命令，会返回一条错误信息。

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

BLACKLISTED_USERS = [
  '12345' # Restrict access for a user ID for a contractor
]

robot.receiveMiddleware (context, next, done) ->
  if context.response.message.user.id in BLACKLISTED_USERS
    # Don't process this message further.
    context.response.message.finish()

    # If the message starts with 'hubot' or the alias pattern, this user was
    # explicitly trying to run a command, so respond with an error message.
    if context.response.message.text?.match(robot.respondPattern(''))
      context.response.reply "I'm sorry @#{context.response.message.user.name}, but I'm configured to ignore your commands."

    # Don't process further middleware.
    done()
  else
    next(done)

Receive Middleware API

Receive中间件的回调函数接受3个参数:context、next和done。

context包含这些字段：

• response
  • response此时并没有match属性，因为listeners还没有被执行所以还没有匹配。
  • 中间件用一些额外的信息来装饰（不是修改）response（比如为response.message.user添加LDAP groups信息）
  • 中间可以修改response.message对象

RESPONSE MIDDLEWARE

Response中间件在hubot发送消息给聊天室的时候执行。对消息格式化，防止密码泄漏，指标等很有用。

Response Middleware Example

示例修改了发送至聊天室的链接。

1
2
3
4
5

module.exports = (robot) ->
  robot.responseMiddleware (context, next, done) ->
    return unless context.plaintext?
    context.strings = (string.replace(/\[([^\[\]]*?)\]\((https?:\/\/.*?)\)/, "<$2|$1>") for string in context.strings)
    next()

Response Middleware API

Response中间件的回调函数接受3个参数:context、next和done。

context包含这些字段：

• response
  • response可以用来在中间件中发送新的消息。中间件会再次执行。小心无限循环。
• strings
  • 发送给聊天适配器的字符串数组。你可以修改这些信息，或者像context.strings = ["new strings"]这样来替换
• method
  • 字符串类型，表示listener发送的消息类型，比如send, reply, emote 或者 topic
• plaintext
  • true或者undefined。如果消息是正常的 plaintext类型，则会设为true。比如send 和reply。该属性只读。

TESTING HUBOT SCRIPTS

hubot-test-helper是用来测试Hubot脚本的很好的测试框架。

安装包：

1

% npm install hubot-test-helper --save-dev

同时还需要安装:

• 一个测试框架，如 Mocha
• 断言库，如 chai 或者 expect.js

可能还需要安装:

• coffee-script (如果你想用CoffeeScript写测试)
• mock的库，比如 Sinon.js (如果你的脚本运行是webservice或者其他异步的服务)

这里是个简单的例子，使用 Mocha, chai, coffee-script 和 hubot-test-helper:

test/example-test.coffee

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

Helper = require('hubot-test-helper')
chai = require 'chai'

expect = chai.expect

helper = new Helper('../scripts/example.coffee')

describe 'example script', ->
  beforeEach ->
    @room = helper.createRoom()

  afterEach ->
    @room.destroy()

  it 'doesn\'t need badgers', ->
    @room.user.say('alice', 'did someone call for a badger?').then =>
      expect(@room.messages).to.eql [
        ['alice', 'did someone call for a badger?']
        ['hubot', 'Badgers? BADGERS? WE DON\'T NEED NO STINKIN BADGERS']
      ]

  it 'won\'t open the pod bay doors', ->
    @room.user.say('bob', '@hubot open the pod bay doors').then =>
      expect(@room.messages).to.eql [
        ['bob', '@hubot open the pod bay doors']
        ['hubot', '@bob I\'m afraid I can\'t let you do that.']
      ]

  it 'will open the dutch doors', ->
    @room.user.say('bob', '@hubot open the dutch doors').then =>
      expect(@room.messages).to.eql [
        ['bob', '@hubot open the dutch doors']
        ['hubot', '@bob Opening dutch doors']
      ]

sample output

1
2
3
4
5
6
7
8
9
10

% mocha --compilers "coffee:coffee-script/register" test/*.coffee


  example script
    ✓ doesn't need badgers
    ✓ won't open the pod bay doors
    ✓ will open the dutch doors


  3 passing (212ms)

使用第三方API的一个常见问题是峰值期间可用性问题。即便是非峰值期间，API请求仍可能被拒绝、超时或者失败。这篇文章中，我将讲解对于RabbitMQ消费者轻量级的指数退避和重试机制。

New API, new service

当用户从我们的一个网站删除了一张照片之后，我们会将它从我们的CDN服务中删除。我们的后端是面向服务的架构，这里使用了CDN purge服务。这个服务对核心web应用发生的事情做出反应，发送消息队列，并且调用CDN的API清除资源。最近我更新了这个服务来使用Akamai CCU REST API。

由于现有的服务已经存在很久了，我们决定重写它。对于一个初级开发者来说，这是一个很好的机会使用RabbitMQ和Node.js来实现后台任务。

正如Paul在他最近的blog中所说RabbitMQ: Front the Front Line，我们是RabbitMQ的重度使用者。尽管我们很多消费者都是实用Ruby来实现的，但是我们发现Node.js在很少的业务逻辑下也是非常实用。事件驱动和消息流很契合，并且适用于快速编码。我们实用node-amqp作为我们的RabbitMQ客户端。

一旦新的服务运行起来，很显然负载会很高，尤其是高峰8pm-2am。我们很快发现，我们的API被集中在晚上调用。这个API有一个内部队列用来处理purge请求，当队列满了的时候它会返回507 queue is full。

观察调用成功和失败的API，很显然我们在高峰时期需要强大的退避和重试机制。

Starting point

我们的平台发送多个不同的消息路由至CDN purge队列。CDN purge服务消费这些消息并且发送对应的请求给CDN API。下图描述了从平台到API的信息流。

当API请求被拒绝的时候，我们想过一段时间再重试。与此同时，我们该如何处理消息？

Let RabbitMQ do the work

实现退避和重试机制，我的第一直觉是创建一个新的wait queue，把失败的请求放在里面，过一段时间继续重试。由于我刚刚使用RabbitMQ，有这么几个问题需要考虑：

• 是否我需要消费者处理wait queue中的消息
• 是否我能控制每个消息的重试等待时间
• 是否我能追踪每个API请求我重试的次数
• 是否能在一个wait queue中处理多个平台的事情

感谢的是，RabbitMQ有很多protocol extensions扩展了AMQP规范。
dead letter exchanges和per-message TTL正是wait queue需要的功能。

Dead letter exchanges (DLX)

术语dead letter mail 死信邮件在邮政行业仍在使用，用来描述信件没有办法被送达。在RabbitMQ中，消息有以下几种情况的时候会被认为dead-lettered

• 消息被拒绝
• 消息已经过期
• 队列已经满了

邮政服务会将死信邮件返还给发送者，和这很相似，RabbitMQ会做一些工作并且重新将dead-lettered的消息发送给我们选择的exchange-dead letter exchange。

既然我们想要一个wait queue，消息过期最容易触发dead-lettering。我们将控制消息短期内过期。

任何队列都能设置dead-letter消息。dead letter exchange是队列的一个参数，当你声明的时候可以通过x-dead-letter-exchange参数来设定。这是一个使用node-amqp的例子：

1
2
3
4
5

var queueOptions = { arguments: { "x-dead-letter-exchange": "exchange" } };

connection.queue("wait-queue", queueOptions, function(waitQueue) {
  // Bind to exchange
});

尽管dead letter exchanges没有做什么特别的配置，我们仍是有了wait queue。下一步我们将设置消息过期，这样RabbitMQ就能为我们重新分配。

更多信息: RabbitMQ docs on DLX

Per-message TTL

消息可以使用默认的expiry或者TTL来声明。然而为了实现指数退避，我们需要对每个消息逐个设置过期时间。

当你发布消息的时候，你可以设置expiration字段的毫秒值：

1
2
3

var messageOptions = { expiration: 10000 };

exchange.publish("routing-key", "body", messageOptions);

看起很简单，但是这也意味着，当一个purge请求失败的时候，因为需要expiration字段，我们的消费者需要复制消息。注意：如果你声明了你的队列使用消息确认机制，不要忘记确认原来的消息。

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

var subscribeOptions = { ack: true };

queue.subscribe(subscribeOptions, function(message, headers, deliveryInfo, messageObject) {
  // Post request to API
  // ...

  // If the API request fails
  var messageOptions = {
    appId: messageObject.appId,
    timestamp: messageObject.timestamp,
    contentType: messageObject.contentType,
    deliveryMode: messageObject.deliveryMode,
    headers: headers,
    expiration: 10000
  };
  exchange.publish(deliveryInfo.routingKey, message, messageOptions);
  messageObject.acknowledge(false);
});

你需要确保你复制你自己的消息的详细信息。下面让我们增加每次API调用失败时的的超时时间。

更多信息: RabbitMQ docs on per-message TTL

Handling dead-lettered messages

当消息dead-lettered之后，RabbitMQ对它做了一些很有用的变化，在header中记录了这些变化。对于我们的wait queue，我们仅仅关心对expiration字段发生了什么。

expiration字段被移除了，并且重新在headers中的x-death中作为original-expiration值记录着。这允许我们找到上次的过期时间并且防止消息再次过期。重要的是，x-deathheader是有序数组，第一个是最近的值。

1
2
3
4
5
6
7
8
9

var expiration;

if (headers["x-death"]) {
  expiration = (headers["x-death"][0]["original-expiration"] * 3);
} else {
  expiration = 10000;
}
// Apply some randomness to the expiration
// ...

这个例子中，第一个个过期时间是10000毫秒，每次重试的时候它都会乘以3。这是指数退避算法中比较常见的实践。我们的例子中，我们增加了一点随即值来提高API调用的成功可能性。

下面我们来组织我们的队列，这样他能管理多个平台的事件。

Routing dead-lettered messages

我们的CDN purge服务对于多个平台发生的事情做出反应，每个都有它自己的routing key。最简单的方式路由多个routing key的是声明单独的wait exchange。

使用单独的wait exchange之后，你可以不用理会routing keys。当失败的消息发送到wait exchange时，你不需要改变routing key。仅仅需要将wait queue和primary queue绑定同样的routing key。

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

var routingKeys = ["routing-key-a", "routing-key-b"];

connection.exchange("wait-exchange", waitExchangeOptions, function(waitExchange) {
  var waitQueueOptions = { arguments: { "x-dead-letter-exchange": "exchange" } };

  connection.queue("wait-queue", waitQueueOptions, function(waitQueue) {
    // Bind wait queue to all routing keys on wait exchange
    routingKeys.map(function(routingKey) {
      waitQueue.bind("wait-exchange", routingKey);
    });
  });
});

connection.exchange("primary-exchange", primaryExchangeOptions, function(primaryExchange) {
  connection.queue("primary-queue", primaryQueueOptions, function(primaryQueue) {
    // Bind primary queue to all routing keys on primary exchange
    routingKeys.map(function(routingKey) {
      primaryQueue.bind("primary-exchange", routingKey);
    });
    // Subscribe to messages
    // ...
  });
});

这样配置之后，当消息被wait queue dead-lettered之后，会重新发布至primary exchange，routing key会保持不变。这样以后添加和移除routing key会很简单。

All together now

现在把所有的代码合并在一起。

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

var amqp = require("amqp");

connection = amqp.createConnection({ host: "localhost" });

connection.on("ready", function() {
  var waitExchange,
      routingKeys = ["routing-key-a", "routing-key-b"];

  waitExchange = connection.exchange("wait-exchange", waitExchangeOptions, function(waitExchange) {
    var waitQueueOptions = { arguments: { "x-dead-letter-exchange": "primary-exchange" } };

    connection.queue("wait-queue", waitQueueOptions, function(waitQueue) {
      routingKeys.map(function(routingKey) {
        waitQueue.bind("wait-exchange", routingKey);
      });
    });
  });

  connection.exchange("primary-exchange", primaryExchangeOptions, function(primaryExchange) {
    connection.queue("primary-queue", primaryQueueOptions, function(primaryQueue) {
      var subscribeOptions = { ack: true };

      routingKeys.map(function(routingKey) {
        primaryQueue.bind("primary-exchange", routingKey);
      });

      primaryQueue.subscribe(subscribeOptions, function(message, headers, deliveryInfo, messageObject) {
        var expiration, messageOptions;
        // Post request to API
        // ...

        // If the API request fails
        if (headers["x-death"]) {
          expiration = (headers["x-death"][0]["original-expiration"] * 3);
        } else {
          expiration = 10000;
        }
        messageOptions = {
          appId: messageObject.appId,
          timestamp: messageObject.timestamp,
          contentType: messageObject.contentType,
          deliveryMode: messageObject.deliveryMode,
          headers: headers,
          expiration: expiration
        };
        waitExchange.publish(deliveryInfo.routingKey, message, messageOptions);
        messageObject.acknowledge(false);
      });
    });
  });
});

Summary

我已经介绍了如何使用dead letter exchanges和per-message TTL来实现轻量级别的指数退避和重试机制。上面的示例代码展示了在Node.js中使用node-amqp如何实现。下图表述了这个机制原理：

如果你和第一个图比较，我希望可以清楚的解释其中原理。最后，下面对开始的问题做一些简要的回答：

是否我需要消费者处理wait queue中的消息?

否。RabbitMQ可以做这个工作，声明一个带x-dead-letter-exchange参数的wait queue，RabbitMQ会在他们过期的时候重新发布。

是否我能控制每个消息的重试等待时间?

可以。但是per-message TTL仅能在你发送消息的时候设置。所以的你的消费者需要复制消息体并且发送的时候需要携带expiration字段。注意：如果使用了消息确认机制，不要忘了确认原来的消息。

是否我能追踪每个API请求我重试的次数?

是。每次消息变为dead-lettered时，RabbitMQ都会记录有用的信息在它的x-deathheader。第一条记录的是最新的，并且会包含original-expiration

是否能在一个wait queue中处理多个平台的事情

是。最简单的管理多个routing key的做法是为你的wait queue声明一个单独的exchange，然后和primary exchange绑定同样的routing key。

原文 Back-off and retry with RabbitMQ

Java Collections Framework是Java提供的对集合进行定义，操作，和管理的包含一组接口，类的体系结构。

Java集合框架的基本接口/类层次结构：

java.util.Collection [I]
+—java.util.List [I]
+—java.util.ArrayList [C]
+—java.util.LinkedList [C]
+—java.util.Vector [C]
+—java.util.Stack [C]
+—java.util.Set [I]
+—java.util.HashSet [C]
+—java.util.SortedSet [I]
+—java.util.TreeSet [C]

java.util.Map [I]
+—java.util.SortedMap [I]
+—java.util.TreeMap [C]
+—java.util.Hashtable [C]
+—java.util.HashMap [C]
+—java.util.LinkedHashMap [C]
+—java.util.WeakHashMap [C]

[I]：接口
[C]：类

Collection接口

Collection是最基本的集合接口，一个Collection代表一组Object的集合，这些Object被称作Collection的元素。

所有实现Collection接口的类都必须提供两个标准的构造函数：无参数的构造函数用于创建一个空的Collection，有一个Collection参数的构造函数用于创建一个新的Collection，这 个新的Collection与传入的Collection有相同的元素。后一个构造函数允许用户复制一个Collection。

如何遍历Collection中的每一个元素？不论Collection的实际类型如何，它都支持一个iterator()的方法，该方法返回一个迭代子，使用该迭代子即可逐一访问Collection中每一个元素。典型的用法如下：

1
2
3
4

Iterator it = collection.iterator(); // 获得一个迭代子 
while(it.hasNext()) { 
　　Object obj = it.next(); // 得到下一个元素 
}

根据用途的不同，Collection又划分为List与Set。

List接口

List继承自Collection接口。List是有序的Collection，使用此接口能够精确的控制每个元素插入的位置。用户能够使用索引（元素在List中的位置，类似于数组下标）来访问List中的元素，这类似于Java的数组。

跟Set集合不同的是，List允许有重复元素。对于满足e1.equals(e2)条件的e1与e2对象元素，可以同时存在于List集合中。当然，也有List的实现类不允许重复元素的存在。

除了具有Collection接口必备的iterator()方法外，List还提供一个listIterator()方法，返回一个 ListIterator接口，和标准的Iterator接口相比，ListIterator多了一些add()之类的方法，允许添加，删除，设定元素， 还能向前或向后遍历。

实现List接口的常用类有LinkedList，ArrayList，Vector和Stack。

LinkedList类

LinkedList实现了List接口，允许null元素。此外LinkedList提供额外的get，remove，insert方法在 LinkedList的首部或尾部。这些操作使LinkedList可被用作堆栈（stack），队列（queue）或双向队列（deque）。

注意LinkedList没有同步方法。如果多个线程同时访问一个List，则必须自己实现访问同步。一种解决方法是在创建List时构造一个同步的List：

1

List list = Collections.synchronizedList(new LinkedList(...));

ArrayList类

ArrayList实现了可变大小的数组。它允许所有元素，包括null。ArrayList没有同步。
size，isEmpty，get，set方法运行时间为常数。但是add方法开销为分摊的常数，添加n个元素需要O(n)的时间。其他的方法运行时间为线性。

每个ArrayList实例都有一个容量（Capacity），即用于存储元素的数组的大小。这个容量可随着不断添加新元素而自动增加，但是增长算法并 没有定义。当需要插入大量元素时，在插入前可以调用ensureCapacity方法来增加ArrayList的容量以提高插入效率。

和LinkedList一样，ArrayList也是非同步的（unsynchronized）。

Vector类

Vector非常类似ArrayList，但是Vector是同步的。由Vector创建的Iterator，虽然和ArrayList创建的 Iterator是同一接口，但是，因为Vector是同步的，当一个Iterator被创建而且正在被使用，另一个线程改变了Vector的状态（例如，添加或删除了一些元素），这时调用Iterator的方法时将抛出

ConcurrentModificationException，因此必须捕获该异常。

Stack 类

Stack继承自Vector，实现一个后进先出的堆栈。Stack提供5个额外的方法使得Vector得以被当作堆栈使用。基本的push和pop方 法，还有peek方法得到栈顶的元素，empty方法测试堆栈是否为空，search方法检测一个元素在堆栈中的位置。Stack刚创建后是空栈。

Set接口

Set继承自Collection接口。Set是一种不能包含有重复元素的集合，即对于满足e1.equals(e2)条件的e1与e2对象元素，不能同时存在于同一个Set集合里，换句话说，Set集合里任意两个元素e1和e2都满足e1.equals(e2)==false条件，Set最多有一个null元素。

因为Set的这个制约，在使用Set集合的时候，应该注意：

1. 为Set集合里的元素的实现类实现一个有效的equals(Object)方法。
2. 对Set的构造函数，传入的Collection参数不能包含重复的元素。

请注意：必须小心操作可变对象（Mutable Object）。如果一个Set中的可变元素改变了自身状态导致Object.equals(Object)=true将导致一些问题。

HashSet类

此类实现 Set 接口，由哈希表（实际上是一个 HashMap 实例）支持。它不保证集合的迭代顺序；特别是它不保证该顺序恒久不变。此类允许使用 null 元素。

HashSet不是同步的，需要用以下语句来进行S同步转换：

1

Set s = Collections.synchronizedSet(new HashSet(...));

Map接口

Map没有继承Collection接口。也就是说Map和Collection是2种不同的集合。Collection可以看作是（value）的集合，而Map可以看作是（key，value）的集合。

Map接口由Map的内容提供3种类型的集合视图，一组key集合，一组value集合，或者一组key-value映射关系的集合。

Hashtable类

Hashtable继承Map接口，实现一个key-value映射的哈希表。任何非空（non-null）的对象都可作为key或者value。

添加数据使用put(key, value)，取出数据使用get(key)，这两个基本操作的时间开销为常数。

Hashtable 通过initial capacity和load factor两个参数调整性能。通常缺省的load factor 0.75较好地实现了时间和空间的均衡。增大load factor可以节省空间但相应的查找时间将增大，这会影响像get和put这样的操作。
使用Hashtable的简单示例如下，将1，2，3放到Hashtable中，他们的key分别是”one”，”two”，”three”：

1
2
3
4

Hashtable numbers = new Hashtable(); 
numbers.put("one", new Integer(1)); 
numbers.put("two", new Integer(2)); 
numbers.put("three", new Integer(3));

要取出一个数，比如2，用相应的key：

1
2

Integer n = (Integer)numbers.get("two"); 
System.out.println("two =" + n);

由于作为key的对象将通过计算其散列函数来确定与之对应的value的位置，因此任何作为key的对象都必须实现hashCode和equals方 法。hashCode和equals方法继承自根类Object，如果你用自定义的类当作key的话，要相当小心，按照散列函数的定义，如果两个对象相 同，即obj1.equals(obj2)=true，则它们的hashCode必须相同，但如果两个对象不同，则它们的hashCode不一定不同，如 果两个不同对象的hashCode相同，这种现象称为冲突，冲突会导致操作哈希表的时间开销增大，所以尽量定义好的hashCode()方法，能加快哈希 表的操作。

如果相同的对象有不同的hashCode，对哈希表的操作会出现意想不到的结果（期待的get方法返回null），要避免这种问题，只需要牢记一条：要同时复写equals方法和hashCode方法，而不要只写其中一个。

Hashtable是同步的。

HashMap类

HashMap和Hashtable类似，不同之处在于HashMap是非同步的，并且允许null，即null value和null key。，但是将HashMap视为Collection时（values()方法可返回Collection），其迭代子操作时间开销和HashMap 的容量成比例。因此，如果迭代操作的性能相当重要的话，不要将HashMap的初始化容量设得过高，或者load factor过低。

WeakHashMap类

WeakHashMap是一种改进的HashMap，它对key实行“弱引用”，如果一个key不再被外部所引用，那么该key可以被GC回收。

对集合操作的工具类

Java提供了java.util.Collections，以及java.util.Arrays类简化对集合的操作

java.util.Collections主要提供一些static方法用来操作或创建Collection，Map等集合。

java.util.Arrays主要提供static方法对数组进行操作。。

总结

如果涉及到堆栈，队列等操作，应该考虑用List，对于需要快速插入，删除元素，应该使用LinkedList，如果需要快速随机访问元素，应该使用ArrayList。

如果程序在单线程环境中，或者访问仅仅在一个线程中进行，考虑非同步的类，其效率较高，如果多个线程可能同时操作一个类，应该使用同步的类。

在除需要排序时使用TreeSet,TreeMap外,都应使用HashSet,HashMap,因为他们 的效率更高。

要特别注意对哈希表的操作，作为key的对象要正确复写equals和hashCode方法。

容器类仅能持有对象引用（指向对象的指针），而不是将对象信息copy一份至数列某位置。一旦将对象置入容器内，便损失了该对象的型别信息。

尽量返回接口而非实际的类型，如返回List而非ArrayList，这样如果以后需要将ArrayList换成LinkedList时，客户端代码不用改变。这就是针对抽象编程。

注意：

1. Collection没有get()方法来取得某个元素。只能通过iterator()遍历元素。
2. Set和Collection拥有一模一样的接口。
3. List，可以通过get()方法来一次取出一个元素。使用数字来选择一堆对象中的一个，get(0)…。(add/get)
4. 一般使用ArrayList。用LinkedList构造堆栈stack、队列queue。
5. Map用 put(k,v) / get(k)，还可以使用containsKey()/containsValue()来检查其中是否含有某个key/value。
   HashMap会利用对象的hashCode来快速找到key。
6. Map中元素，可以将key序列、value序列单独抽取出来。
   使用keySet()抽取key序列，将map中的所有keys生成一个Set。
   使用values()抽取value序列，将map中的所有values生成一个Collection。
   为什么一个生成Set，一个生成Collection？那是因为，key总是独一无二的，value允许重复。

assigned

主键由外部程序负责生成，在 save() 之前必须指定一个。Hibernate不负责维护主键生成。与Hibernate和底层数据库都无关，可以跨数据库。在存储对象前，必须要使用主键的setter方法给主键赋值，至于这个值怎么生成，完全由自己决定，这种方法应该尽量避免。

1
2
3

<id name="id" column="id">
<generator class="assigned" />
</id>

“ud”是自定义的策略名，人为起的名字，后面均用“ud”表示。

特点：可以跨数据库，人为控制主键生成，应尽量避免。

increment

由Hibernate从数据库中取出主键的最大值（每个session只取1次），以该值为基础，每次增量为1，在内存中生成主键，不依赖于底层的数据库，因此可以跨数据库。

1
2
3

<id name="id" column="id">
<generator class="increment" />
</id>

Hibernate调用org.hibernate.id.IncrementGenerator类里面的generate()方法，使用select max(idColumnName) from tableName语句获取主键最大值。该方法被声明成了synchronized，所以在一个独立的Java虚拟机内部是没有问题的，然而，在多个JVM同时并发访问数据库select max时就可能取出相同的值，再insert就会发生Dumplicate entry的错误。所以只能有一个Hibernate应用进程访问数据库，否则就可能产生主键冲突，所以不适合多进程并发更新数据库，适合单一进程访问数据库，不能用于群集环境。

官方文档：只有在没有其他进程往同一张表中插入数据时才能使用，在集群下不要使用。

特点：跨数据库，不适合多进程并发更新数据库，适合单一进程访问数据库，不能用于群集环境。

hilo

hilo（高低位方式high low）是hibernate中最常用的一种生成方式，需要一张额外的表保存hi的值。保存hi值的表至少有一条记录（只与第一条记录有关），否则会出现错误。可以跨数据库。

1
2
3
4
5
6
7
8
9
10

<id name="id" column="id">
<generator class="hilo">
<param name="table">hibernate_hilo</param>
<param name="column">next_hi</param>
<param name="max_lo">100</param>
</generator>
</id>
<param name="table">hibernate_hilo</param> 指定保存hi值的表名
<param name="column">next_hi</param> 指定保存hi值的列名
<param name="max_lo">100</param> 指定低位的最大值

也可以省略table和column配置，其默认的表为hibernate_unique_key，列为next_hi

1
2
3
4
5

<id name="id" column="id">
<generator class="hilo">
<param name="max_lo">100</param>
</generator>
</id>

hilo生成器生成主键的过程（以hibernate_unique_key表，next_hi列为例）：

1. 获得hi值：读取并记录数据库的hibernate_unique_key表中next_hi字段的值，数据库中此字段值加1保存。
2. 获得lo值：从0到max_lo循环取值，差值为1，当值为max_lo值时，重新获取hi值，然后lo值继续从0到max_lo循环。
3. 根据公式 hi * (max_lo + 1) + lo计算生成主键值。

注意：当hi值是0的时候，那么第一个值不是0*(max_lo+1)+0=0，而是lo跳过0从1开始，直接是1、2、3……

那max_lo配置多大合适呢？

这要根据具体情况而定，如果系统一般不重启，而且需要用此表建立大量的主键，可以吧max_lo配置大一点，这样可以减少读取数据表的次数，提高效率；反之，如果服务器经常重启，可以吧max_lo配置小一点，可以避免每次重启主键之间的间隔太大，造成主键值主键不连贯。

特点：跨数据库，hilo算法生成的标志只能在一个数据库中保证唯一。

seqhilo

与hilo类似，通过hi/lo算法实现的主键生成机制，只是将hilo中的数据表换成了序列sequence，需要数据库中先创建sequence，适用于支持sequence的数据库，如Oracle。

1
2
3
4
5
6

<id name="id" column="id">
<generator class="seqhilo">
<param name="sequence">hibernate_seq</param>
<param name="max_lo">100</param>
</generator>
</id>

特点：与hilo类似，只能在支持序列的数据库中使用。

sequence

采用数据库提供的sequence机制生成主键，需要数据库支持sequence。如oralce、DB、SAP DB、PostgerSQL、McKoi中的sequence。MySQL这种不支持sequence的数据库则不行（可以使用identity）。

1
2
3
4

<generator class="sequence">
<param name="sequence">hibernate_id</param>
</generator>
<param name="sequence">hibernate_id</param> 指定sequence的名称

Hibernate生成主键时，查找sequence并赋给主键值，主键值由数据库生成，Hibernate不负责维护，使用时必须先创建一个sequence，如果不指定sequence名称，则使用Hibernate默认的sequence，名称为hibernate_sequence，前提要在数据库中创建该sequence。

特点：只能在支持序列的数据库中使用，如Oracle。

identity

identity由底层数据库生成标识符。identity是由数据库自己生成的，但这个主键必须设置为自增长，使用identity的前提条件是底层数据库支持自动增长字段类型，如DB2、SQL Server、MySQL、Sybase和HypersonicSQL等，Oracle这类没有自增字段的则不支持。

1
2
3

<id name="id" column="id">
<generator class="identity" />
</id>

例：如果使用MySQL数据库，则主键字段必须设置成auto_increment。

1

id int(11) primary key auto_increment

特点：只能用在支持自动增长的字段数据库中使用，如MySQL。

native

native由hibernate根据使用的数据库自行判断采用identity、hilo、sequence其中一种作为主键生成方式，灵活性很强。如果能支持identity则使用identity，如果支持sequence则使用sequence。

1
2
3

<id name="id" column="id">
<generator class="native" />
</id>

例如MySQL使用identity，Oracle使用sequence

注意：如果Hibernate自动选择sequence或者hilo，则所有的表的主键都会从Hibernate默认的sequence或hilo表中取。并且，有的数据库对于默认情况主键生成测试的支持，效率并不是很高。

使用sequence或hilo时，可以加入参数，指定sequence名称或hi值表名称等，如

1

<param name="sequence">hibernate_id</param>

特点：根据数据库自动选择，项目中如果用到多个数据库时，可以使用这种方式，使用时需要设置表的自增字段或建立序列，建立表等。

uuid

UUID：Universally Unique Identifier，是指在一台机器上生成的数字，它保证对在同一时空中的所有机器都是唯一的。按照开放软件基金会(OSF)制定的标准计算，用到了以太网卡地址、纳秒级时间、芯片ID码和许多可能的数字，标准的UUID格式为：

xxxxxxxx-xxxx-xxxx-xxxxxx-xxxxxxxxxx (8-4-4-4-12)

其中每个 x 是 0-9 或 a-f 范围内的一个十六进制的数字。

1
2
3

<id name="id" column="id">
<generator class="uuid" />
</id>

Hibernate在保存对象时，生成一个UUID字符串作为主键，保证了唯一性，但其并无任何业务逻辑意义，只能作为主键，唯一缺点长度较大，32位（Hibernate将UUID中间的“-”删除了）的字符串，占用存储空间大，但是有两个很重要的优点，Hibernate在维护主键时，不用去数据库查询，从而提高效率，而且它是跨数据库的，以后切换数据库极其方便。

特点：uuid长度大，占用空间大，跨数据库，不用访问数据库就生成主键值，所以效率高且能保证唯一性，移植非常方便，推荐使用。

guid

GUID：Globally Unique Identifier全球唯一标识符，也称作 UUID，是一个128位长的数字，用16进制表示。算法的核心思想是结合机器的网卡、当地时间、一个随即数来生成GUID。从理论上讲，如果一台机器每秒产生10000000个GUID，则可以保证（概率意义上）3240年不重复。

1
2
3

<id name="id" column="id">
<generator class="guid" />
</id>

Hibernate在维护主键时，先查询数据库，获得一个uuid字符串，该字符串就是主键值，该值唯一，缺点长度较大，支持数据库有限，优点同uuid，跨数据库，但是仍然需要访问数据库。

注意：长度因数据库不同而不同

MySQL中使用select uuid()语句获得的为36位（包含标准格式的“-”）

Oracle中，使用select rawtohex(sys_guid()) from dual语句获得的为32位（不包含“-”） 特点：需要数据库支持查询uuid，生成时需要查询数据库，效率没有uuid高，推荐使用uuid。

foreign

使用另外一个相关联的对象的主键作为该对象主键。主要用于一对一关系中。

1
2
3
4
5
6

<id name="id" column="id">
<generator class="foreign">
<param name="property">user</param>
</generator>
</id>
<one-to-one name="user" class="domain.User" constrained="true" />

该例使用domain.User的主键作为本类映射的主键。

特点：很少使用，大多用在一对一关系中。

select

使用触发器生成主键，主要用于早期的数据库主键生成机制，能用到的地方非常少。

其他注释方式配置

注释方式与配置文件底层实现方式相同，只是配置的方式换成了注释方式

1. 自动增长，适用于支持自增字段的数据库

   1 2	@Id @GeneratedValue(strategy = GenerationType.IDENTITY)

2. 根据底层数据库自动选择方式，需要底层数据库的设置
   如MySQL，会使用自增字段，需要将主键设置成auto_increment。

   1 2	@Id @GeneratedValue(strategy = GenerationType.AUTO)

3. 使用表存储生成的主键，可以跨数据库。
   每次需要主键值时，查询名为”hibernate_table”的表，查找主键列”gen_pk”值为”2”记录，得到这条记录的”gen_val”值，根据这个值，和allocationSize的值生成主键值。

   1 2 3 4 5 6 7 8 9

   @Id @GeneratedValue(strategy = GenerationType.TABLE, generator = "ud") @TableGenerator(name = "ud", table = "hibernate_table", pkColumnName= "gen_pk", pkColumnValue = "2", valueColumnName = "gen_val", initialValue = 2, allocationSize = 5)

4. 使用序列存储主键值

   1 2 3 4 5 6

   @Id @GeneratedValue(strategy = GenerationType.SEQUENCE, generator = "ud") @SequenceGenerator(name = "ud", sequenceName = "hibernate_seq", allocationSize = 1, initialValue = 2)

5. uuid生成主键

   1 2 3

   @Id @GenericGenerator(name = "systemUUID", strategy = "uuid") @GeneratedValue(generator = "systemUUID")

小结

1. 为了保证对象标识符的唯一性与不可变性，应该让Hibernate来为主键赋值，而不是程序。

2. 正常使用Hibernate维护主键，最好将主键的setter方法设置成private，从而避免人为或程序修改主键，而使用assigned方式，就不能用private，否则无法给主键赋值。

3. Hibernate中唯一一种最简单通用的主键生成器就是uuid。虽然是个32位难读的长字符串，但是它没有跨数据库的问题，将来切换数据库极其简单方便，推荐使用！

4. 关于hilo机制注意：
   hilo算法生成的标志只能在一个数据库中保证唯一。
   当用户为Hibernate自行提供连接，或者Hibernate通过JTA，从应用服务器的数据源获取数据库连接时，无法使用hilo，因为这不能保证hilo单独在新的数据库连接的事务中访问hi值表，这种情况，如果数据库支持序列，可以使用seqhilo。

5. 使用identity、native、GenerationType.AUTO等方式生成主键时，只要用到自增字段，数据库表的字段必须设置成自动增加的，否则出错。

6. 还有一些方法未列出来，例如uuid.hex，sequence-identity等，这些方法不是很常用，且已被其他方法代替，如uuid.hex，官方文档里建议不使用，而直接使用uuid方法。

7. Hibernate的各版本主键生成策略配置有略微差别，但实现基本相同。如，有的版本默认sequence不指定序列名，则使用名为hibernate_sequence的序列，有的版本则必须指定序列名。

8. 还可以自定义主键生成策略，这里暂时不讨论，只讨论官方自带生成策略。