股票行情数据是一种典型的时序数据（Time-series Data），在一般的IT系统中，日志数据其实也是一种时序数据，在大数据的世界里，也有大量应用是基于时序数据处理的，可以说时间序列的数据无处不在。所以，哪怕是不炒股、不熟悉金融世界的工程师，从本文也可以了解一些具有普遍性的技术思考，例如在大规模、高密度的数据处理中，是把数据快照搬到计算节点作运算还是把计算能力放到数据节点中“就地计算”？协程（co-routine）在这类计算密集型系统中又有何作用？

实时行情服务是券商的基础服务，给普通投资者描绘出风云变幻的动态市场画面，也给量化投资者提供最重要的建模基础数据和下单信号，时间就是金钱，行情服务必须要快。证券交易系统行情指标很丰富，最基础常见的包括：行情报价、分笔数据、分时数据、分钟K线、日周月年K线、各类财务技术指标、多维度排序、多维度统计等等，这些指标需要由交易所的行情数据流结合时间流进行统计计算。本文分享广发证券行情服务并行化计算演进的过程，包括五个部分：

• 股票行情是怎么回事

• 日均10亿次的行情指标计算

• 基于Redis+Lua的方案：“就地计算”

• 引入Goroutine的方案：“海量算子”

• 孰优孰劣

一、不炒股没关系，股票行情科普在这里

任何交易在达成之前，通常都有一个讨价还价的僵持过程，或者买方让价，或者卖方让价，两不相让的时候需要第三方介入撮合取个“平均价”，这个过程就是定价。证券股票交易也不例外，不同的是买卖双方互不可见，双方通过券商渠道把自己的价量报给交易所，交易所通常按照达成最大成交量的原则撮合定价。撮合涉及order book和tick data两个概念。交易所维护两个“账本”分别记录买方和卖方申报的价量，实时或按照一定频率进行撮合定价成交，申报、成交时对“账本”进行增、改、删操作。

这两个“账本”就称作order book，对order book增改删操作引起数据变化，每个变化的快照就称作tick data。国内股民数超 1.2 亿户，A股上市公司近3000家，可以想见的到order book是一个大“账本”，tick data瞬息万变，一个交易日产生的tick data量更是惊人。国内交易所目前采用抓取快照的方式，抓取order book的前五/十档价量，剔除”无用的”其他档价量数据，统计交易当日开市时点到目前时点的最高、最低、成交量、成交额数据，我们将这种数据称为原始行情数据（如图1）。

交易所将原始行情数据近实时的发送给券商，券商行情系统对这份原始数据进一步处理，比如按时间区间统计出5分钟、10分钟、15分钟、30分钟、1小时、1天、1周、1月、1季度、1年这十个周期时间段K线蜡烛图的高、开、低、收价格及成交量、成交额数据。

券商将处理好的行情数据揭示给投资者做再报价参考。可见这类数据量大、变化频繁、中间统计计算耗时，一旦延迟，以后就再也跟不上“实时”的节奏了。又快又准，是股票行情服务的基本要求，任何误差、延迟，都可以引起交易者的经济损失，导致投诉甚至社会事件。系统该如何设计才能高速的处理和展示这类实时行情数据，是个很大的技术挑战。

图1

二、你们股炒的爽，我们指标算的酸爽

券商行情系统可以分为行情原始数据接收解析、中间指标统计计算处理、行情数据请求响应三个模块。原始数据接收解析模块只需要做好与不同交易所的行情消息格式适配即可，行情数据请求响应模块要达到原始行情报价及各类行情统计指标的快速展现需要减少中间处理步骤，在行情原始数据接收解析、统计指标计算完成后立即推送给终端用户，同时存一份到内存数据库中以便终端用户再次查询读取，推送与查询两路相结合，达到在数据落地之前即已展示给用户的效果。

剩下要做的就是要尽量减少中间指标统计计算的处理时延。A股上市公司3000多家，基金加债券数更是上万，但它们之间是无关联的，在统计计算时完全可以分片、并行处理，存储上采用内存数据库，redis内置多种数据结构的支持满足多统计指标存储的需求、容易分片部署便于并行计算，如图2。

图 2

以最简单的十个周期时间段的K线统计指标，证券数保守算1w只，10*1w=10w，也就是说分分秒秒就有近10w的计算量，国内沪深交易所一个交易日开市4个小时，按照交易所行情数据每3秒更新一次，可得出日计算量就有4.8亿，加上其他指标计算如市盈率、涨跌幅、换手率、委比委差、多板块多指标排序，日计算量已突破10亿。

三、“就地计算”的方案 – 把计算能力放在数据里

好在不同的证券可以分片并行、同样各周期段的K线指标也可以并行统计计算，实时处理这么大的计算量，我们显然需要高度并行处理。

我们最早的方案基于Redis，因为Redis是一个非常好的承载时间序列数据的高性能内存存储技术。Redis除了内置多种数据结构，还内置了Lua语言解释器，这意味着Redis除了具备数据存储能力也拥有了数据计算的能力。一个Redis存储集群中，每个节点加载Lua脚本，这基本上就是一个分布式并行计算集群了。

我们剩下要做的是事情是实现一个行情收集器，接收来自不同交易所的行情原始数据（node.js善于处理io型事务，很适合用来实现收集器，细节不是本文焦点，不在此详述），发送eval lua指令到Redis集群。 Redis收到指令后执行我们用Lua实现的指标计算逻辑完成指标计算，之后通过Redis pub将行情数据推送出去。如图3通过将计算挪到Redis存储节点，我们避免了复杂易出错的多进程管理问题，也大大简化了开发的工作量。

图 3

这个方案的一个优点，是技术架构比较简单，从数据存储到运算处理，都在Redis上，我们仅专注于Redis集群的性能优化、高可用方案实现、容灾备份、数据复制。对于运维来说，运维一套相对单一的技术系统，“零部件”（moving parts）越少越好，出现故障、单点失败的环节也少了很多。

四、“海量算子”- 把数据快照挪到技术节点

上述Redis+Lua的方案，早期也服务了我们的市场与客户，但是当指标计算量不断增大后，其不足也日益显著，主要体现在高度密集的计算导致影响存储集群的性能而产生延迟，并且在交易期间对存储集群作动态扩容是非常困难的。

有鉴于此，我们很自然的只能把指标计算任务从数据存储中回收，放弃一个数据与计算一体化的相对简单的架构，把计算的职责交给存储之外的专用运算节点。此时Redis变成单纯的内存存储。这种实现将计算与存储分离，计算所需的数据不再能从”本地”获取到了，对于Redis而言，运算服务算是“out of process”（进程外），所以计算指标所需的数据，必须以一份数据快照的方式从Redis传递到运算节点，用以计算和更新，计算结果最终回写到Redis，如图4。

这个对于Redis而言“out of process”的计算能力，我们称之为运算节点（相对于Redis的数据节点），是一系列非常容易水平扩容的、高度并发的程序，我们采用了Golang来实现。Golang这个语言，天生支持协程(Goroutine) – 在一个运行的Golang程序中，可轻易启动上万的协程，所消耗的资源远小于线程，天生适合并行计算。

当接收到交易所tick数据时， 在一个运算节点中对每只证券解析及每类指标计算启动单独的goroutine，每个goroutine在它的生命周期中只做一件事就结束（存活时间毫秒级），这很像数学上的函数执行过程完全没有副作用，goroutine就是一个闭包算子，相互之间毫无影响。 

Golang的内存回收是并行的，数万个goroutine启动到销毁对性能的影响很小，另外tick data本身是有间隔的从交易所发过来的，在goroutine销毁那一刻往往是tick data空闲的间隔时间，这个空闲时机点用来做goroutine回收是合适的。当然也有一些常驻goroutine用来将指标结果数据落地到Redis。程序语言本身是有各自不同的设计哲学的，Golang正是这样一种语言，其协程机制让我们能够更细粒度的处理可分而治之的系统，同时免去了复杂易出错的多进程、多线程问题。

图 4

五、方案比较 - 孰优孰劣

两种方案的对比：

1、Redis既负责存储也负责计算

• 我们通过搭建Redis集群来进行分布式并行计算，Redis集群本身有主备节点，这就涉及到主备同步的问题，虽然Redis自己解决了同步问题并且也支持增量同步，但通过eval lua指令在存储节点上计算时，同步的不是计算结果而是计算本身，也就是说同一个指标计算在主节点和备节点都需要各自计算一次。Redis又是单线程处理，对于复杂的指标计算通过查看SLOWLOG会有上百毫秒的延迟，这会造成阻塞，对于要求快速的行情服务是不可接受的，为了减少阻塞，就需要尽可能多的进行分片，这意味着需要起更多的redis节点。

• Redis集群一旦搭建，节点缩扩容需要人工干预。对于证券行情服务这种目前9:30-15:00业务高峰，其他时段空闲的系统来说，高峰时无法弹性扩容，低峰时Redis节点进程无法回收造成资源浪费。

2、Redis负责存储，海量Goroutine做算子

• 将指标计算从Redis拿出来交给goroutine，Redis仅作存储节点，主备节点同步的是指标计算结果，这样降低了Redis进程的cpu使用率，也不再有SLOWLOG记录，同时仅需要很少的分片，大大减少了Redis节点数。

• Redis集群规模的大小仅需要根据业务数据量确定。指标计算量的变化可以轻易通过启动更多的goroutine来进行方便的弹性扩容，goroutine数也随投资者活跃程度变化，对交易频繁的股票只需要启动更多的goroutine，不会像方案1那样造成部分Redis节点成为热点。在休市时段goroutine完全回收释放。

图 5

从Redis+Lua的数据存储与运算一体化，过度到数据存储与并行运算分离，也大大减少了系统对硬件资源的要求，Redis集群规模减少十倍，如图表1。采用Redis+Lua方案实现上比较简单，开发工作量较小，收集器与Redis数据交换量小，这种方案更适合简单的逻辑计算比如计数器；goroutine的实现方案，虽然开发上复杂，与Redis数据交换量大了一些，但更适合像行情指标计算这类复杂的应用场景。

   如果说用传统关系数据库如MSSQL,MYSQL,PGSQL来设计一个行情数据库，除表名外，就是列字段了。

       比如，表名：600036.SH_1min,  (这里假定每个个股设计一张表，避免把所有的个股放在一张表中，导致数据以亿计，查询效率太低)

       举例而言，其中一条记录（虚拟）

        code :600036.SH Date :2014-10-09 DateTime: 930 Open: 9.60 High:10.08 Close:10.02 Low :9.54 Volume: 123456 Amount :78964531 Factor: 5.123

        另外，成份股的指数权属作为另外一张表而存在，不放在这里，否则太大，也浪费空间。

        如果以内存数据库Redis来设计key-value,相对的key 就会比较长，但value比较简单。

         因为key的唯一性，

         比如，同样以上面的记录而言，可能就要设计一个唯一的KEY:

         key1:  600036.SH_1min:2014-10-09:0930:Open  => value1 :9.60

         key2:  600036.SH_1min:2014-10-09:0930:High    => value 2:10.08

         key3:  600036.SH_1min:2014-10-09:0930:Close  => value3 :10.02

         key4:  600036.SH_1min:2014-10-09:0930:Low     => value4 :9.54

          ......

          也就是说

          set   600036.SH_1min:2014-10-09:0930:Open  9.60

          set   600036.SH_1min:2014-10-09:0930:High  10.08

          set   600036.SH_1min:2014-10-09:0930:Close   10.02

          set   600036.SH_1min:2014-10-09:0930:Low  9.54

           .......

          这样，就可以建立一系列的KEY-VALUE结构的非结构化数据了。

         下面，针对历史行情数据库中，典型的日度数据，N分钟数据，TICK数据，我们要进行一个整体上的数据库设计。

         (1) 类型的设计

              我个人认为，用hash的方式来设历史行情库，可能会更好。

         (2) KEY 的设计

              一般而言，而行情数据库的KEY要包含代码名称（比如，600036.SH）.但是，因为分钟数据和日度数据的粒度相差较大，如何统一考虑？

             其中，不同类型的数据的文件名应有唯一性。

              A、分钟数据：

                           文件变量名："600036.sh_1min“，假设有6年的数据，6*250*270  约40万条左右的数据记录源（注意KEY个数还要乘以记录的字段数，下同。）

                            KEY ： “600036.sh_1min_2014-10-09_930”  => 开盘价，收盘价，最高价，最低价，成交量，成交金额，复权因子....

              B、TICK数据：

                           文件变量名：”600036.sh_Tick_2014-10-09“  

                                                  #  把每天的TICK，有数千条（股票）甚至数万条（期货）放到一个变量中。

                           KEY： ”600036.sh_Tick_2014-10-09 _930_01.001“ =>

              C、日度数据：

                          文件变量名：”600036.sh_1day“，假设6年数据，此文件大约会储存 6*250 =1500条左右数据源。

                          KEY： “ 600036.sh_1day_2014-10-09” =>

             另外，数据中应考虑不同类类型的数据，可能具有不同的特征，如 股票：复权因子，期货：未平仓合约量 。

        （3）性能比较

             数据如何设计，关键要看性能和维度容易程度。这块需要后续实证了。

nginx + lua + redis 防刷和限流

 redis的常用命令主要分为两个方面、一个是键值相关命令、一个是服务器相关命令

在golang的项目中，若要频繁的用redis（或者其他类似的NoSQL）来存取数据，最好用redigo自带的池来管理连接。

不然的话，每当要操作redis时，建立连接，用完后再关闭，会导致大量的连接处于TIME_WAIT状态（redis连接本质上就是tcp）。

注：TIME_WAIT，也叫TCP半连接状态，会继续占用本地端口。

以下为redis连接池的golang实现：

import (

"github.com/garyburd/redigo/redis"

"github.com/astaxie/beego"

"time"

)

var (

// 定义常量

RedisClient     *redis.Pool

REDIS_HOST string

REDIS_DB   int

)

func init() {

// 从配置文件获取redis的ip以及db

REDIS_HOST = beego.AppConfig.String("redis.host")

REDIS_DB, _ = beego.AppConfig.Int("redis.db")

// 建立连接池

RedisClient = &redis.Pool{

// 从配置文件获取maxidle以及maxactive，取不到则用后面的默认值

MaxIdle:     beego.AppConfig.DefaultInt("redis.maxidle", 1),

MaxActive:   beego.AppConfig.DefaultInt("redis.maxactive", 10),

IdleTimeout: 180 * time.Second,

Dial: func() (redis.Conn, error) {

c, err := redis.Dial("tcp", REDIS_HOST)

if err != nil {

return nil, err

}

// 选择db

c.Do("SELECT", REDIS_DB)

return c, nil

},

}

}

其中，各参数的解释如下：

MaxIdle：最大的空闲连接数，表示即使没有redis连接时依然可以保持N个空闲的连接，而不被清除，随时处于待命状态。

MaxActive：最大的激活连接数，表示同时最多有N个连接

IdleTimeout：最大的空闲连接等待时间，超过此时间后，空闲连接将被关闭

Dial：建立连接

使用连接池时的代码：

// 从池里获取连接

rc := RedisClient.Get()

// 用完后将连接放回连接池

defer rc.Close()

以上就是连接池的用法了，很简单吧。

GraphicsMagick号称图像处理领域的瑞士军刀。提供了健壮及高效的图像处理工具包和库，支持超过88种主流图片格式包括：BMP，GIF，JPEG，JPEG-2000，PNG，PDF，PNM，TIFF，DPX…

现在最新稳定版本为：1.3.12。安装之前，因为是图片处理，所以需要系统中安装了libpng和libjpeg的开发包，否则的话不会安装这两种文件的支持。

官方下载地址：http://www.graphicsmagick.org/<br><br><br>http://www.imagemagick.org/download/<br><br><br>

命令格式：gm convert [ options ... ] input_file output_file

下面给出一些常用玩法，本文也将陆续添加其他玩法，敬请关注！

重定义尺寸，取样质量

gm convert -quality 80 -resize 100×100 input.jpg output.jpg

加文字水印，指定字体、字体大小、颜色、位置

gm convert -font ArialBold -pointsize 45 -fill red -draw “text 100,100 www.saysth.com” input.jpg output.jpg

加图片水印至右下角，透明度50%

gm composite -gravity southeast -dissolve 50 watermark.png input.jpg output.jpg

加图片水印至制定位置，透明度50%

 gm composite -geometry +50+50 -dissolve 50 watermark.png input.jpg output.jpg<br><br><br>http://www.imagemagick.org/download/

 time包中的Add和Sub的用法，Add用于计算某个时间之前和之后的时间点，Sub用于计算两个时间差

package main

import (

"fmt"

"strings"

"time"

)

func main() {

// Add 时间相加

now := time.Now()

// ParseDuration parses a duration string.

// A duration string is a possibly signed sequence of decimal numbers,

// each with optional fraction and a unit suffix,

// such as "300ms", "-1.5h" or "2h45m".

//  Valid time units are "ns", "us" (or "µs"), "ms", "s", "m", "h".

// 10分钟前

m, _ := time.ParseDuration("-1m")

m1 := now.Add(m)

fmt.Println(m1)

// 8个小时前

h, _ := time.ParseDuration("-1h")

h1 := now.Add(8 * h)

fmt.Println(h1)

// 一天前

d, _ := time.ParseDuration("-24h")

d1 := now.Add(d)

fmt.Println(d1)

printSplit(50)

// 10分钟后

mm, _ := time.ParseDuration("1m")

mm1 := now.Add(mm)

fmt.Println(mm1)

// 8小时后

hh, _ := time.ParseDuration("1h")

hh1 := now.Add(hh)

fmt.Println(hh1)

// 一天后

dd, _ := time.ParseDuration("24h")

dd1 := now.Add(dd)

fmt.Println(dd1)

printSplit(50)

// Sub 计算两个时间差

subM := now.Sub(m1)

fmt.Println(subM.Minutes(), "分钟")

sumH := now.Sub(h1)

fmt.Println(sumH.Hours(), "小时")

sumD := now.Sub(d1)

fmt.Printf("%v 天\n", sumD.Hours()/24)

}

func printSplit(count int) {

fmt.Println(strings.Repeat("#", count))

}

 以前总是通过timestamp来防止重放攻击，但是这样并不能保证每次请求都是一次性的。今天看到了一篇文章介绍的通过nonce（Number used once）来保证一次有效，感觉两者结合一下，就能达到一个非常好的效果了。

重放攻击是计算机世界黑客常用的攻击方式之一，所谓重放攻击就是攻击者发送一个目的主机已接收过的包，来达到欺骗系统的目的，主要用于身份认证过程。

首先要明确一个事情，重放攻击是二次请求，黑客通过抓包获取到了请求的HTTP报文，然后黑客自己编写了一个类似的HTTP请求，发送给服务器。也就是说服务器处理了两个请求，先处理了正常的HTTP请求，然后又处理了黑客发送的篡改过的HTTP请求。

基于timestamp的方案

每次HTTP请求，都需要加上timestamp参数，然后把timestamp和其他参数一起进行数字签名。因为一次正常的HTTP请求，从发出到达服务器一般都不会超过60s，所以服务器收到HTTP请求之后，首先判断时间戳参数与当前时间相比较，是否超过了60s，如果超过了则认为是非法的请求。

假如黑客通过抓包得到了我们的请求url： 
http://koastal.site/index/Info?uid=ZX07&stime=1480862753&sign=80b886d71449cb33355d017893720666 
其中

$sign=md5($uid.$token.$stime); // 服务器通过uid从数据库中可读出token
1
2

一般情况下，黑客从抓包重放请求耗时远远超过了60s，所以此时请求中的stime参数已经失效了。 
如果黑客修改stime参数为当前的时间戳，则sign参数对应的数字签名就会失效，因为黑客不知道token值，没有办法生成新的数字签名。

但这种方式的漏洞也是显而易见的，如果在60s之后进行重放攻击，那就没办法了，所以这种方式不能保证请求仅一次有效。

基于nonce的方案

nonce的意思是仅一次有效的随机字符串，要求每次请求时，该参数要保证不同，所以该参数一般与时间戳有关，我们这里为了方便起见，直接使用时间戳的16进制，实际使用时可以加上客户端的ip地址，mac地址等信息做个哈希之后，作为nonce参数。 
我们将每次请求的nonce参数存储到一个“集合”中，可以json格式存储到数据库或缓存中。 
每次处理HTTP请求时，首先判断该请求的nonce参数是否在该“集合”中，如果存在则认为是非法请求。

假如黑客通过抓包得到了我们的请求url： 
http://koastal.site/index/Info?uid=ZX07&nonce=58442c21&sign=80b886d71449cb33355d017893720666

其中

$sign=md5($uid.$token.$nonce); // 服务器通过uid从数据库中可读出token
1
2

nonce参数在首次请求时，已经被存储到了服务器上的“集合”中，再次发送请求会被识别并拒绝。 
nonce参数作为数字签名的一部分，是无法篡改的，因为黑客不清楚token，所以不能生成新的sign。

这种方式也有很大的问题，那就是存储nonce参数的“集合”会越来越大，验证nonce是否存在“集合”中的耗时会越来越长。我们不能让nonce“集合”无限大，所以需要定期清理该“集合”，但是一旦该“集合”被清理，我们就无法验证被清理了的nonce参数了。也就是说，假设该“集合”平均1天清理一次的话，我们抓取到的该url，虽然当时无法进行重放攻击，但是我们还是可以每隔一天进行一次重放攻击的。而且存储24小时内，所有请求的“nonce”参数，也是一笔不小的开销。

基于timestamp和nonce的方案

那我们如果同时使用timestamp和nonce参数呢？ 
nonce的一次性可以解决timestamp参数60s的问题，timestamp可以解决nonce参数“集合”越来越大的问题。

我们在timestamp方案的基础上，加上nonce参数，因为timstamp参数对于超过60s的请求，都认为非法请求，所以我们只需要存储60s的nonce参数的“集合”即可。

假如黑客通过抓包得到了我们的请求url： 
http://koastal.site/index/Info?uid=ZX07&stime=1480862753&nonce=58442c21&sign=80b886d71449cb33355d017893720666

其中

$sign=md5($uid.$token.$stime.$nonce); // 服务器通过uid从数据库中可读出token
1
2

如果在60s内，重放该HTTP请求，因为nonce参数已经在首次请求的时候被记录在服务器的nonce参数“集合”中，所以会被判断为非法请求。超过60s之后，stime参数就会失效，此时因为黑客不清楚token的值，所以无法重新生成签名。

综上，我们认为一次正常的HTTP请求发送不会超过60s，在60s之内的重放攻击可以由nonce参数保证，超过60s的重放攻击可以由stime参数保证。

因为nonce参数只会在60s之内起作用，所以只需要保存60s之内的nonce参数即可。

我们并不一定要每个60s去清理该nonce参数的集合，只需要在新的nonce到来时，判断nonce集合最后一次修改时间，超过60s的话，就清空该集合，存放新的nonce参数集合。其实nonce参数集合可以存放的时间更久一些，但是最少是60s。

验证流程

//判断stime参数是否有效 if( $now - $stime > 60){     die("请求超时"); } //判断nonce参数是否在“集合”已存在 if( in_array($nonce,$nonceArray) ){     die("请求仅一次有效"); } //验证数字签名     if ( $sign != md5($uid.$token.$stime.$nonce) ){     die("数字签名验证失败"); } //判断是否需要清理nonce集合 if( $now - $nonceArray->lastModifyTime > 60 ){     $nonceArray = null; } //记录本次请求的nonce参数 $nonceArray.push($nonce);   //开始处理合法的请求