学习东西有三个境界：了解、熟练使用、理解原理。大神一般都是第四个境界：根据原理造轮子。对于RabbitMQ，我目前大概在第二个境界。以下介绍均按照自己理解来，个人不保证完全正确，而且也认为肯定有些不准或错误的地方，建议仅作参考，欢迎指正。

一、RabbitMQ是一款开源企业级消息代理软件，实现了AMQP协议(和MQTT协议)。

AMQP，即Advanced Message Queuing Protocol,一个提供统一消息服务的应用层标准高级消息队列协议。
MQTT，即Message Queuing Telemetry Transport，是一种轻量级的、灵活的网络协议，常用于物联网。
官网：https://www.rabbitmq.com/
文档：https://www.rabbitmq.com/documentation.html

二、主要概念

1. Virtual Host:
   你可以参照windows概念：一机多用户。为了在一个单独的代理上实现多个隔离的环境（用户、用户组、交换机、队列 等），AMQP提供了一个虚拟主机（virtual hosts - vhosts）的概念。Virtual Host是隔离的最高级别。

2. Exchange: 交换机
   交换机用来发送AMQP实体消息。接收生产者的消息，将消息路由到[0, n](n>0,整数)个队列。消息内容可以是文本，也可以是二进制数据。不同类型的交换机有不同的消息路由策略。每个队列创建的时候，必须指定一个Exchange，这个Exchange是下面描述中的某一种: (以下中文名可能和网上的不一样)

   2.1 Direct exchange: 直连交换机。
   根据消息携带的路由键(routing key)将消息投递给对应队列(完全匹配)，可以发送给多个队列(消息复制)。

   2.2 Fanout exchange: 扇形交换机。
   将消息发给绑定到该交换机的所有队列，忽略(队列绑定到exchange时绑定的)路由键。可以处理广播消息。

   2.3 Topic exchange: 主题交换机。

   a. 根据消息携带的路由键，将消息发送给相应队列。和Direct不同的是，主题交换机的匹配规则可以是模糊匹配。
   
   b. "#"(井号)表示任意数量(0个or多个)单词。当一个队列的绑定键为 "#"（井号） 的时候，这个队列将会无视消息的路由键，接收所有的消息。
   
   c. "*"(星号)表示一个单词。当 * (星号) 和 # (井号) 这两个特殊字符都未在绑定键中出现的时候，此时主题交换机就拥有的直连交换机的行为。
   

   2.4 Headers exchange: 消息头交换机/首部交换机。

   a. 和http请求一样，每个RabbitMQ的消息也有消息头，消息头内部有很多键值对，这些键值对可以自己定义。
   
   b. 头交换机使用一个或多个消息头(x-match可以为any或all)来代替路由键建立路由规则。通过判断消息头的值能否与指定的绑定相匹配来确立路由规则。
   
   c. x-match为any时，任意一个消息头符合即可。相应地，all表示必须满足所有定义的消息头。
   

3. queue: 队列
   队列需要连接到某具体的交换机(在代码中, channel创建时需要指定交换机，queue再通过channel发送/接收消息，channel是建立在tcp连接上的可复用逻辑通道，避免频繁的tcp连接)。

4. channel: 通道
   4.1 见queue中的描述。
   4.2 channel可以减少tcp频繁创建、断开的开销。
   4.3 binding_key: 绑定
   4.4 在绑定（Binding）Exchange与Queue的同时，一般会指定一个binding key；消费者将消息发送给Exchange时，一般会指定一个routing key；当binding key与routing key相匹配时，消息将会被路由到对应的Queue中。
   4.5 routing_key在某些交换机下会被忽略，相应地binding_key也会被忽略(binding_key的作用就是和routing_key做匹配)。
   4.6 可以理解成binding_key就是生产者指定的"routing_key"。

5. 消息消费模式
   5.1 订阅-发布
   5.2 消费者主动拉取

6. 部分特性
   6.1 消息持久化。消息持久化需要以下指标同时持久化：

   a. 交换机持久化
   
   b. 队列持久化
   
   c. 消息设置持久化(Delivery Mode => 2)
   
   d. 如果消息、队列设置了过期时间，过期后的消息会被发送到“死信队列”(如果设置了的话)，否则丢弃。
   
   e. 注意上面一条，持久化的消息也可能因为设置过期时间而被丢弃。
   

   6.2 过期时间说明：Message的TTL("expiration")是从数据到达Queue中后开始计时的，而不是Message被创建时计时。Queue的TTL("x-expires")定义是Queue被自动删除前可以处于未使用状态的时间。如果你设置了Queue为自动删除，那么"x-expires"这个参数就有效。

   官方解释：
   [1] Auto-delete：queue that has had at least one consumer is deleted when last consumer unsubscribes

   [2] Expiry time can be set for a given queue by setting the x-expires argument to queue.declare, or by setting the expires policy. This controls for how long a queue can be unused before it is automatically deleted. Unused means the queue has no consumers

   一条消息什么时候过期是由两个因素决定的：Message中的"expiration"，以及Queue中的"x-message-ttl"，他们都是一种消息过期策略，消息具体过期时间以 min("x-expires", "x-message-ttl")为准。

   6.3 消费者ack机制:

   a. 在确认消息已被获取后，才删除消息；
   
   b. 自动确认ack机制: 如果消息不重要，自动ack的消息可以增加消息的消费速度。
   

   6.4 生产者confirm机制：

   a. 确认消息成功投递（到队列）后，执行回调函数；
   
   b. 不论是持久化，还是生产者confirm，对RabbitMQ吞吐量都有一定影响；
   

背景

对于比较关键的下单接口，性能直接影响到商户的使用体验。目前，我们的下单接口大部分耗时都在4s-5s之间，少数耗时在2-3s。具体耗时多少，也和实际覆盖到的流程有关，不同分支流程，需要处理的业务逻辑往往不一样，调用接口数量也不一样。

思路

4s-5s耗时算比较多的，考虑到有些商户网络比较差，客户端从下单到接收到返回数据的实际耗时会更多，实际体验也会更差。

缩减耗时，可以从mysql优化入手，但是我们的sql已经挺快了（我的意思是，它的优化空间并不是很大，继续优化的难度会越来越大），网络耗时的优化也难以有很明显的效果。

从ztrace的监控面板上可以看到，几乎所有请求都是串联的，只要改为并发，就可以大大减小接口总耗时。这么多的接口，只要拿5、6个做并发，就可以节省接近1s的时间，所有优化空间很大。

特别是一些诸如“获取用户信息”、“获取商户信息”、“获取商品信息”之类的接口，它们对执行顺序没有特别要求，完全可以在接口开始时就做并发。

调用链跟踪系统的一个巨大优点就是直观，每个部分的耗时都可以显示出来，并且进行对比。

操刀

PHP有对并发访问的支持：curl_multi_init，但是要用到我们项目里，要尽量减小对原流程的干涉，所以做了一些独特的设计。

cURL批处理的用法，是先初始化一个批处理句柄：curl_multi_init，然后通过curl_multi_add_handle把普通的curl句柄加入批处理句柄，再统一执行(并发)。

下面是一个例子，来自：php.net

<?php
// 创建一对cURL资源
$ch1 = curl_init();
$ch2 = curl_init();

// 设置URL和相应的选项
curl_setopt($ch1, CURLOPT_URL, "http://www.example.com/");
curl_setopt($ch1, CURLOPT_HEADER, 0);
curl_setopt($ch2, CURLOPT_URL, "http://www.php.net/");
curl_setopt($ch2, CURLOPT_HEADER, 0);

// 创建批处理cURL句柄
$mh = curl_multi_init();

// 增加2个句柄
curl_multi_add_handle($mh,$ch1);
curl_multi_add_handle($mh,$ch2);

$running=null;
// 执行批处理句柄
do {
    usleep(10000);
    curl_multi_exec($mh,$running);
} while ($running > 0);

// 关闭全部句柄
curl_multi_remove_handle($mh, $ch1);
curl_multi_remove_handle($mh, $ch2);
curl_multi_close($mh);

?>

我直接找了一个相关库，并把它clone到了公司github上，保留了授权（感谢Apache License，感谢作者，附上原地址：jmathai/php-multi-curl，内部项目地址：shop/php-multi-curl）

为了少改业务流程，我设计了一个方法：preExec，它的参数和execApi一样。
它的作用是生成请求的curl句柄，并将其加入到批处理(还没执行)，然后计算请求的hash，设置$ci->preExec[$request_hash](可以通过它去获取这次请求的结果)。

如何计算请求hash：请求hash的计算元素，包括请求url、请求params、请求method。

function requestHash($url, $p, $method)
{
    $url = strtolower(trim(trim($url), '&?/\\'));
    $phash = $this->arrayHash($p);
    $method = strtolower(trim($method)) == 'post' ? 'post' : 'get';
    return md5($url . $phash . $method);
}
function arrayHash($p, $filter = array('mdstr', 'expire'))
{
    $para_sort = array();
    foreach ($p as $k => $v) {
        if ($filter && in_array($v, $filter, true)) continue;
        // array object bool int float null resource...
        $para_sort[$k] = is_array($v) ? $this->arrayHash($v)
            : (is_object($v) ? $this->arrayHash((array)$v) : strval($v));
    }
    ksort($para_sort);
    return md5(http_build_query($para_sort));
}

当我们去主动获取请求结果时，如果当前请求已经执行完成，会立即返回请求结果，其他还没执行完的请求继续执行，不堵塞业务流程。如果当前请求没执行完，会堵塞直到当前请求执行结束。

执行execApi时，先判断是否有设置$ci->preExec[$request_hash]，如果没设置，就是一次普通的curl访问，和以前的流程一样(单次请求，堵塞直到获取到结果)；如果有设置，那就执行curl批处理（所有的并发请求都在这时开始执行。如果已经执行过了就不再执行），并通过$ci->preExec[$request_hash]去获取当前请求的结果。获取到结果后，会将$ci->preExec[$request_hash]删除。

所以，只要我们在流程开始前执行preExec，那执行到execApi的时候就会自动做并发，并获取结果。

在流程1和流程2，execApi分别执行普通curl访问、并发curl访问。

DEBUG流程

下面讲一些遇到的坑。

开始做并发时很顺利，也成功实现了并发。但是当我把它和调用链跟踪结合起来的时候，一大堆问题就来了。

为了准确跟踪接口调用时间，我们需要记录每个curl的执行开始时刻，和执行过程的耗时。

每创建一个curl句柄后，我们都会向curl批处理会话中添加这个单独的curl句柄，但是这个时候curl是没开始执行的。如果我们把创建curl的时刻作为url开始访问的时刻，记录到的接口生命周期会比实际的长很多(提前记录了)。

这时我想到了通过记录curl结束时间和curl耗时，反推开始时间。而curl有个回调函数，可以在接收到数据的时候执行：CURLOPT_WRITEFUNCTION(有没有curl开始执行的回调函数呢？很遗憾没有！)

于是，我使用了CURLOPT_WRITEFUNCTION来记录curl调用的结束时刻。但是在235上，设置CURLOPT_WRITEFUNCTION后总是得不到返回数据（curl_exec总是返回TRUE），我发现CURLOPT_RETURNTRANSFER设置失效了！

** 235上的libcurl版本是：7.19.7**

CURLOPT_WRITEFUNCTION 回调函数名。该函数应接受两个参数。第一个是 cURL resource；第二个是要写入的数据字符串。数 据必须在函数中被保存。 函数必须准确返回写入数据的字节数，否则传输会被一个错误所中 断。

CURLOPT_RETURNTRANSFER TRUE 将curl_exec()获取的信息以字符串返回，而不是直接输出。

一顿折腾，在Stack Overflow上找到一点提示 ready go

When I placed that line before any of the other options, it was simply ignored.

我尝试在设置CURLOPT_WRITEFUNCTION后再次设置CURLOPT_RETURNTRANSFER，这次跑通了。

BUT！我自己的电脑上又出问题了，设置了CURLOPT_WRITEFUNCTION，CURLOPT_RETURNTRANSFER就失效，CURLOPT_RETURNTRANSFER生效的时候，就不执行CURLOPT_WRITEFUNCTION。在需要二选一的时候，我选择了放弃。

** 我电脑上的libcurl版本是：7.55.0**

我放弃了CURLOPT_WRITEFUNCTION。找到了一个替代回调函数CURLOPT_HEADERFUNCTION，这个函数会在接收到返回的header信息时调用。虽然它不能精确表示结束时间，但是马马虎虎也能用。最重要的是，这个函数始终都会被调用。

CURLOPT_HEADERFUNCTION 设置一个回调函数，这个函数有两个参数，第一个是cURL的资源句柄，第二个是输出的 header 数据。header数据的输出必须依赖这个函数，返回已写入的数据大小。

这时，调用链跟踪和并发成功结合起来了，监控的数据基本正确，也能正常运行。

如果看到ztrace上面，一个并发前只需要100ms+的接口，在并发后要800ms+，请相信我，curl_getinfo($ch, CURLINFO_TOTAL_TIME)返回的就是这个值，不是程序有bug。具体原因期待大神解释。

调试过程中还遇到一个libcurl的bug，这个bug只在 7.55.0 上有（没错，我电脑上的就是这个版本:broken_heart:）。这个bug就是curl_getinfo($ch, CURLINFO_TOTAL_TIME)会在某些时候返回一个超大的浮点数，比如4295.xx。而经过我观察，只需要减掉4295就行了（究竟对不对，就看我猜的对不对了:pray:）

结语

好了，上面是修改ApiShopClient的整个过程，主要时间花在了适配调用链跟踪上面，从这个debug流程也能看到，php在涉及到底层的时候有点力不从心（大神忽略掉这句）。

背景资料

Dapper，大规模分布式系统的跟踪系统

Zipkin

如果要查看zipkin资料，官网的英文文档是最正确的，讲解也很系统；

在阅读文章前，建议先至少看一下Dapper的翻译；

一、介绍

调用链跟踪系统进行了比较大的升级，主要改变有：

1. 对接zipkin日志标准(仍保留了些许不同)；
2. 使用zipkin的数据库和界面；
3. 大幅降低对正常业务流程的性能影响；
4. 将数据压缩后再传递给收集器；
5. rabbitmq使用单独的数据通道，不再和业务共用一个；
6. 采用文件日志和mysql记录结合的方式，分别记录调用链的业务数据和性能数据。 业务数据100%记录，性能数据采样记录(5%)；
7. 支持业务数据的上下文透明传输

下面详细讲解系统的构成、设计考量、使用和特点。

二、结构

浅绿色的是数据最终落地的地方。

名词解释：临时存储

临时存储究竟是干嘛的呢？在zipkin里，一个Span被分割成两部分，一个是Client/Producer，另一个是Server/Consumer。Span的含义更接近于一次RPC，它横跨了两个服务节点，所以在两个节点分别上报信息，最后再合并成一个Span。

后端收到两部分Span的时间不一样，临时存储就用来存储先到达的Span，等另一半到达后再取出来做合并。

名词解释：收集器

从MQ接收上报的Span，做Span的合并，并存储到持久存储(mysql)

名词解释：压缩

压缩成二进制，再通过MQ传递。实际应用中，是先将Span转换为数组，将key替换成a-z的单字母，再转成json后使用zlib压缩（需要zlib扩展）。

和压缩相关的还有一个叫 msgpack 的东西，但是它需要单独安装PHP扩展，所以放弃了。

三、设计考量

考量一：降低对业务的影响

只在少数地方嵌入跟踪代码。实际操作中，在项目入口(index.php)、rpc请求、输出这三个地方做了埋点。

耗时操作在请求以外执行。我们知道PHP没有异步，但是PHP有个非常有用的函数register_shutdown_function，我们把耗时操作全放里面，这样可以尽量减少对线上业务的影响。如果是fastcgi模式，我们会优先使用fastcgi_finish_request这个函数。http://php.net/manual/zh/function.register-shutdown-function.php#108212

考量二：开关/采样率/消息生产速率控制

一共设计了两个开关（业务日志开关、Span上报开关），默认情况下这两个开关都是开启的。在此基础上，增加了采样率控制，仅当Span上报开关开启，且当前Span需要采样时，Span才会通过MQ传递给后端。业务日志开关控制biz-ztrace日志的记录。

需要注意的时，即便把开关全部关闭，traceid也会生成、传递，但不会有日志记录，也不会有Span上报。

对于消息速率控制，在实际中采用的是“过量丢弃”，阈值是10000。只要队列里的消息超过这个值，即不再接收消息，直接抛弃。

考量三：动态增加消费能力

随着系统扩大，采集的数据增多，动态增强数据的消费能力非常重要。目前的做法是通过加开消费者来实现。之所以能直接增加消费者，是因为它们使用的同一个临时存储，我们只是把消息的消费瓶颈，转移到了临时存储的IO瓶颈上。

想象一下，我们有两个消费者A、B，分别使用两个不同的临时存储，Client端的Span被A接收到，Server端的Span被B接收到，由于它们的临时存储隔开的，A、B都无法做Span的合并！

几乎所有的调用链跟踪系统实现方案，都绕不开这个问题，可能是一个Span被分发到不同处理实例，也可能是同一Trace的不同Span被分发到不同处理实例——这都会带来很多麻烦。

解决方法也很多，比如根据traceid做hash，再转发给不同的后端实例，这样同一个trace下的Span都会被，且只会被转发给同一处理实例。

但是，仅仅做hash是不够的，还必须借助“一致性Hash”算法。下面的例子解释了原因。

假设已经有一个处理实例CA，spanA、SpanB都属于同一个trace。CA已经接收到spanA了，这时我们新增了一个CB（和CA使用不同临时存储），如果hash结果表明spanB应该转发给CB（这完全可能发生），那又会出现CA、CB都无法合并这个Span的情况。

关于一致性Hash，我们后面再讲。

考量四：业务日志、性能日志分开

之所以要分开，是因为业务日志很多(随着采集端的增加，数据增长更多)，我们还没处理过这么多的数据。

目前，业务日志记录了$_GET、$_POST、$_SERVER和apiReturn，收集端有seller和api。每天产生的日志共2.5G左右，每月日志总量大概在25G。日志总量和采集端数量呈正相关。

四、使用

下面的例子都有判断$trace_open;，并且使用了 try {} catch () {}，这样做是为了减少可能出现的异常，必须要添加。

记录业务日志

业务日志是记录在文件里的，用的最多。

过期数据处理

我们把太久远的数据称为“过期数据”，因为它们被再次使用的概率很低。我们用的最多的是近期的数据，近期数据才是最重要的，业务日志和性能统计数据都适用这个道理。

我们处理过期数据的方式很简单——打包压缩、删除，这是目前能找到的最划算的处理方法。具体处理逻辑：

文件日志：保留一个月的压缩日志，保留一周内的原始日志。
数据库：保留两个月的数据。

global $trace_open;
if ($trace_open) {
    try {
        $tracer = \Tricolor\ZTracker\Core\GlobalTracer::tracer();
        $tracer->log('get', $_GET);
    } catch (\Exception $e) {}
}

记录跟踪日志

跟踪日志会出现在ui里(如果被采样了的话)

global $trace_open;
if ($trace_open) {
    try {
        $tracer = \Tricolor\ZTracker\Core\GlobalTracer::tracer();
        $tracer->currentSpan()->putTag('store_id', (int)$_POST['store_id']);
    } catch (\Exception $e) {}
}

透明传输业务字段（试验性质）

在调用链上游设置一个值，在调用链下游可以获取到。

之所以说是试验性质，是因为设计里有，也有实现，但实际业务里还没使用过。

global $trace_open;
if ($trace_open) {
    try {
        $tracer = \Tricolor\ZTracker\Core\GlobalTracer::tracer();
        $tracer->currentContext()->set('from', 'mamagou');
        // get
        $tracer->currentContext()->get('from');
    } catch (\Exception $e) {}
}

五、特点

1. 和zipkin的Span比起来，我们的Span结构略有不同。zipkin里的Sampled字段，在我们这里对应Decision字段，Decision字段除了记录是否采样，还记录了log开关和report开关。

2. 数据库、UI使用zipkin，减少了一些开发量。zipkin的Span设计能适应较多的采集场景，比如单向数据传递、黑盒模式的远端节点(调用数据库时，数据库就是一个黑盒模型)。

3. 和淘宝类似，加了业务字段透传功能，业务日志和跟踪日志分开。

问题：

尝试还原数据库，之后提示 table doesn`t exist。

分析：

类型：MyISAM
数据：Table.frm，Table.MYD，Table.MYI
位置：/data/$databasename/目录中
说明：直接复制到mysql中data目录中，便可以使用

类型：InnoDB
数据文件：存储在/$innodb_data_home_dir/中的ibdata1文件中
结构文件：结构文件存在于/data/table_name.frm中
说明：不可以直接使用，并报错：table doesn`t exist

解决方法：

1、 停止 apache 和 mysql服务
2、 拷贝相应文件到/data/目录，在数据库引擎类型为InnoDB时，拷贝数据文件的同时还需要拷贝ibdata1。
3、 将根目录下的ib_logfile*文件全部删除掉

备注：

1、 正常的数据导出恢复，最好用工具，不要在data文件层面去恢复
2、 测试环境在windows下

参考文档：

1、 mysql 直接从date 文件夹备份表，还原数据库之后提示 table doesn`t exist的原因和解决方法

背景

对于无关紧要的进程，我们可以随时kill掉再重启，但是有时候不能这样粗暴地操作。

假设我们正在处理一个从消息系统得到的消息，而消息是 非持久化/自动ack 的，如果进程被打断，这条消息就可能丢失。

或者我们在处理一个耗时特别长的任务，这时别人不小心kill掉了我们的进程，那我们前面的成果就付诸东流了。

针对上面这些情况，我们在Api进程里加入了平滑关闭的特性，你可以捕获进程关闭指令，自己决定何时关闭。

下面是一些使用方法：

使用

使用方法一：闭包

class Cront extends MY_Controller
{
    public function run()
    {
        $ready_exit = false;
        $this->regSig(SIGTERM, function() use (&$ready_exit) {
            $ready_exit = true;
            echo "准备退出！";
        });
        while (true) {
            {
                /**
                 * 括号里是业务处理代码
                 */
                echo "正在处理业务...\n";
                sleep(3);
                echo "业务处理完毕!\n";
            }
            $this->catchSig();
            if ($ready_exit) {
                echo "exit now!\n";
                exit;
            }
        }
    }
}

使用方法二：对象方法

class Cront extends MY_Controller
{
    public $ready_exit = false;
    public function handSig($sig, $my_var1, $my_var2)
    {
        switch ($my_var1) {
            case 'run':
                $this->ready_exit = true;
                echo "exit ready\n";
                break;
        }
        var_dump(func_get_args());
    }

    public function run()
    {
        $this->regSig(SIGTERM, array($this, 'handSig'), 'run', 'myvar');
        while (true) {
            {
                /**
                 * 括号里是业务处理代码
                 */
                echo "正在处理业务...\n";
                sleep(3);
                echo "业务处理完毕!\n";
            }
            $this->catchSig();
            if ($this->ready_exit) {
                echo "exit now!\n";
                exit;
            }
        }
    }
}

注意

1. 进程捕获关闭指令后，用 kill -9 仍然可以杀死进程;
2. api里的 regSig 不仅可以捕获 SIGTERM 信号，还可以捕获其他信号;
3. 如果调用了 regSig 而没有调用 catchSig ，程序仍然捕获不到信号;
4. 不用 declare(ticks = 1); 是考虑到性能;