上个月曾经提过一个高负荷服务器点对点通讯模块的若干问题，目前这个模块已经基本稳定，其中一些朋友的留言对改进这个模块也很有启发。最近主要考虑了通讯故障的问题并进行了测试。

服务器之间的通讯通常会由于网络原因造成各种故障，如果在一个LAN之内，出现故障的概率很小，几乎可以不用考虑。但如果分布在不同的地点比如跨机房跨地域，短时故障的概率就会变大，由于中间线路，路由器等各种设备或者VPN的问题，容易出现短期(比如10-30S左右)对方不可到达的情形。如果每秒系统传递的包有成千上万，那问题就会变得很严重，大量的业务数据投递会发生问题，并且按先后顺序发生以下事件。

(Figure 1 应用程序,操作系统及相关参数的影响)

1. Socket send buffer 未占满期间socket会继续可写, 如果设置了SO_SNDBUF则会使用SO_SNDBUF, 否则使用Linux默认的配置。

# increase TCP max buffer size setable using setsockopt()
  net.core.rmem_max = 16777216
  net.core.wmem_max = 16777216
  # increase Linux autotuning TCP buffer limits
  # min, default, and max number of bytes to use
  # set max to at least 4MB, or higher if you use very high BDP paths
  net.ipv4.tcp_rmem = 4096 87380 16777216 
  net.ipv4.tcp_wmem = 4096 65536 16777216

SO_SNDBUF通常按以下公式设置：BDP = link_bandwidth * RTT
远端
BDP = 100MBps * 0.050 sec / 8 = 640KB
本地
BDP = 1024MBps * 0.005 sec / 8 = 655KB

2. Socket占满后写socket会block, 或者设置了TCP_NODELAY则不管SOCKET buffer是否占满都会block

3. 如果应用写入操作没有队列概念则应用程序会出现异常，所有操作会阻塞。

4. 强壮的应用会对发送数据做应用层的buffer，像Connection Manager/Openfire之间是使用发送 Thread Pool，在对方不可到达的时段内，内存占用会急剧上升。

5. 如果网络层恢复正常，首先是socket buffer中的数据会被发送，然后应用层堆积的数据也随后发送。

6. 如果网络层长时间不可用，有2种方法可以判断，通过达到SO_SNDTIMEO Socket返回错误检测，应用层如果需要更早知道错误，可以调低SO_SNDTIMEO。另外一种检测方法是应用检测发送队列达到上限临界值来做进一步 处理。避免服务器内存溢出造成崩溃。实际环境中需要结合这2种方式一起考虑。

Update: 感兴趣朋友可了解后续文章：多服务器通讯层应该如何设计—一次code review小记

每个大型系统都需要一个完备的监控体系，比如大宝(sodme)近期就在他的关于网游服务器运营的一些基础设施中提到

一个正规的，商业的网游服务器，我想，最起码具备以下几方面的基础设施：

１.程序内的性能，安全，内存等的监控体系．
这套体系，不仅仅包括了日志记录，也包括了故障定位，紧急报警等相应的运营支持……

我在几个月前整理 google talk server 架构介绍 中也有类似描述。

7. 在服务器增加profile/monitor机制，包括配置文件，资源状况, 日志, 可以做离线分析。
这样某个服务器发现问题，首先去看 monitor console
profile/monitor不是侵入式的，不是通过在服务器程序嵌入代码实现。

大型系统通常有多台服务器组成，所以监控体系通常是分布式汇报，统一处理，而不用人工去每台服务器查看一下。而这个“分布式汇报，统一处理”的流程目前业界并没有非常成熟的工具来自动完成。而最近发现 facebook 开源的 Scribe 已经实现了这个功能，由于 facebook 每天要 log 上百亿条级别的信息(也就是每秒log上百万条的信息!)，因此它对于这样一个工具的需求比其他公司更为强烈，也没有其他现成的产品能够满足它这种疯狂的需求，因此它只好自己做了一个。

Scribe 对于中心服务器没有high availability的依赖，当中心服务器不可用时候，所有的信息是log在本地，当中心服务器可用之后，本地保存的信息又会同步到服务器上。

Scribe 本身是由 non-blocking C++ 实现，由于它用的是 facebook 另外一个 Thrift 框架， Thrift 在我的理解就是一个高效的跨语言 rpc 框架。所以它可以支持通过 C++, Java, Python, PHP, Ruby, Erlang, Perl, Haskell, C#, Cocoa, Smalltalk, and OCaml 给服务器汇报。但Scribe下载后的包本身带的例子和配置基本上都是 Python 的。其他语言怎么调用可能还需要自行摸索一下。

Thrift 是优点，但毕竟还是要下一步花时间熟悉一下才会使用。如果 Scribe 是 RESTful 的话，我想我的Scribe系统已经可以跑起来了。

服务器集群之间的多个服务器之间通讯通常有组播或点对点直连发送两种方式。组播可以参看前文JGroups 简介,适用场合,配置,程序例子Demo等完全使用指南。最近由于业务需要，研究了点对点直连的模式，及需要考虑的各种情况。

点对点传输可以用UDP或TCP实现。究竟用什么实现每个人都有自己的喜好。我是坚决 站在TCP这边。我们目前有个程序由于某种原因需要用UDP传输，优化了好几周目前可以达到5-6M/s了(100M LAN)。那TCP可以达到多少，随便叫个高中生写个Socket C/S程序在上面相同的环境100M LAN，可以达到 100Mbps / 8 = 12.5 M/s。

相对于服务器之间通讯来说，传输速率只是一方面，需要考虑的点包括：

• 发送方连接数控制。熟悉Socket的朋友都清楚，Socket通讯简单的方法是发送方用1个固定连接发送(比如SMTP/POP3等)，或发送方每个请求数据包新建一个连接发送(比如PHP/Ruby连MySQL)。对于服务器之间的通讯程序情况可能会比较特殊，通常会出现峰值，比如有几秒服务器负荷大，可能要几秒内传输数万个数据包。采用1个固定连接或每个数据包新建多个连接都有问题。
• 接收方要快。可能有多个发送方同时发过来数据。同时发送的量可能非常大，每秒需要处理数万个并发的数据包。
• 网络或接收方暂时故障。比如服务器之间专线短期故障或者接收方程序崩溃导致不能发送。通常大家自己实现的程序没有 MySQL 或者 Memcached 那么稳定，进程挂掉是概率很高的随机事件。死掉的解决方法可用一个守护进程监测，发现死掉再次重启。那监测到死机到重启需要一个过程，发送方这段事件需要发送的数据怎么处理？
• 如果接收方一直不起来怎么办，发送方会不会被待发送的数据撑满到内存溢出。
• 通讯协议是自定义的，可采用定长包或变长包。如果采用变长包，接收方收到一个坏数据也没法判断这个坏数据有多长怎么办。是不是整个数据流都废了？

所以最近搞的这个东西，95%的功能上面高中生那个程序搞定了，剩下这5%的功能才让人头大。目前大部分细节已经分别处理，简单流程如下。

• 发送方用一个连接池。那多个线程对有限的几个固定连接资源占用怎么分配？可以用发送队列，我现在是用发送线程池。
• 接收方也启用线程池，同时配合non-blocking非阻塞io的方式。
• 暂时失败的数据放入一个队列定时重发。同时记录发送失败的错误次数。
• 如果接收方一直不起来，有业务需要的数据进行持久保存。
• 对于坏数据，最简单把连接close，让发送方再次开新连接发送。

当然这里没有描述全部情况，实际需要考虑的问题比这些更复杂，并且还有不少运行期间新问题待捕捉。

Mina是一种Java中使用广泛的网络框架，Mina写一个Hello world可能5分钟就可以了(参看 Mina测试程序)。 其中最复杂的部分可能就是Decode解码。其作用是将网络的二进制数据流转换成业务逻辑的对象。我前面那个例子为了简化目的，codec用的是Mina中现成的 TextLineDecoder，有些朋友一字不改拿去之后说程序有问题。没错，肯定有问题，因为TextLineDecoder每次读一行， 如果服务器一直不给换行就一直等在那了。那怎么改啊，自己想去吧。

doDecode的返回参数是不太直观的，这两天碰到问题了。我们有个服务器之间通讯的程序，压力测试时候反应很慢，现象有：

• size<1k的包每秒才传几百个左右。但是把包的大小改成8k又能发七八千个。
• 直接在 doDecoder最前面打印，每秒也只有1-200个。
• 怀疑是 send/recv buffer不够大，设成16-64M，没有明显改善。
• tcpdump 查看网络流量，因为信息量太大，也无法得到精确的数字，只是初略>判断包都投递到目标，不是发送方的问题。

    /**
     * Implement this method to consume the specified cumulative buffer and
     * decode its content into message(s).
     *
     * @param in the cumulative buffer
     * @return true if and only if there's more to decode in the buffer
     *         and you want to have doDecode method invoked again.
     *         Return false if remaining data is not enough to decode,
     *         then this method will be invoked again when more data is cumulated.
     * @throws Exception if cannot decode in.
     */
    protected abstract boolean doDecode(IoSession session, ByteBuffer in,
            ProtocolDecoderOutput out) throws Exception;

按上面的意思我是这样实现的,核心片断:

int pos = in.position();
try {
       Object obj = in.getObject();
       out.write(obj);
       return !in.hasRemaining();
} catch (Exception e) {
       in.position(pos);
       return false;
}

问题就在 return !in.hasRemaining(); 上面。这里是应该返回true的。

• 如果你能够解析一次，那就需要返回 true; position 移到解析了的这个末尾
• 如果1个不够解析，position移到最初的位置，返回false
• 解析了一个，后面还有内容，可能是半个，也返回 true

Mina的结构可以参看这里 Openfire (Wildfire) 的最新特性, 基于 Mina 的 Connection Manager

如果你对 ProtocolDecoder 感兴趣，我建议新手去看 ObjectSerializationDecoder / ObjectSerializationEncoder 的代码，特别简洁。不要看 TextLineDecoder, 无关细节太多。

继续改进部分基于用户好友关系的分布式PubSub系统。

本来的想法是把LocalSubList放到远端，好处是High availability即某个节点死了不会造成大面积故障，由另外一个节点可迅速接管。但是如果放在远端, 那更新的开销会非常大, 1个改变需要修改n份远程数据，比如拿在线好友列表说事，一个用户改变状态，所有在线好友的列表需要同时update，在设计上人为制造了一个瓶颈。

因此今天把思路调整下，由发送方广播1个通知, 接收方各自维护更新自己的列表。那HA怎么办，每2个节点分成一组互相替对方存一份。反正发送方会广播通知的，即使增加一个专用的节点来做备份listener也不是什么大问题。在可靠性面前，硬件成本微不足道。

自己做一个网络模块的优点是风险可控，时间可控，但是原理通过之后就只剩下枯燥的开放了。按照don't re-invent the wheel的思想，或许将来更好的做法是底层(语言和框架)来做这些事情。比如网络间的通讯，节点迁移，HA, 数据共享等。

这个模块有点抽象，尽管我觉得不复杂，但跟别人口述的时候对方也是听得一头雾水，从软件项目管理的角度来说设计要尽量简单，至少要项目小组里面一半以上的人一看就明白，因此这个设计还有很大的简化空间。

做软件设计不能做得象编程之美的题目那样高深。对于一个网络服务端程序，我觉得
1. 性能优先
2. 简单优先

需要一个pubsub的功能，用在基于各种好友关系的场合。

* publish list 可能成千上万、十万、百万。
* publish topic 生命周期可能极短，调用一次就结束；也可能很长
* publish 数据实时广播即可，无需保存等待consumer到来
* subscribe list 可能很长，大的数千，也可能很小，只有1个
* subscribe list 相对固定(在线好友列表 or follow list)
* subscribe list 需要跨节点的，即一个topic在多个节点有local subscribe list
* 对性能要求极高，性能为王
* 无事务要求，特殊状况下，如某节点发生故障，丢失小量数据可容忍。
* 分布式，无中心节点
* 节点可动态切换

目前还没找到适合我的现成产品。前几天提到的rabbitmq和erlang或许是一个思路。

Erlang太高深了，周末的时候想了一个适合各种小白语言的思路，试画了一个简单的。

前几个月曾经做个一次 比较Java中几种数据cache方式 的试验，最近看到 Openfire 中有一个非常小巧的本地 Cache实现， 在相同环境测试比流行的ehcache快大约5倍。简单介绍如下。

原理图

实现方法
* 用 HashMap 来存储和用来做 CacheKey 查找。
* 用一个LinkedList来存储访问顺序列表
* 用一个LinkedList来存储添加时间顺序列表，即过期时间。
* HashMap 中 Key 为 CacheKey, Value 包装成一个CacheObject
* CacheObject 包含：
1) object size
2) 指向 Access List 节点的指针
3) 指向 Age List 节点的指针

其中两个List的作用
1) AccessList
当添加新元素且 List 满时，删除列表最后的元素，即最长时间没有访问的元素。

2) AgeList
当调用 get cache 时候，判断 List 末尾有无过期元素，如有向前一直删除到最后一个没有过期的元素为止。

Performance 性能评测
写了个简单的测试，2线程写 Cache, 4 线程同时读Cache，每个Cache 100字节，平均速度大致为

写cache: 168,924 条/秒
读cache: 605,212 次/秒

结果在相同环境测试比流行的ehcache快大约5倍。

Resource资源下载
DefaultCache 源代码，稍修改去掉没用的引用即可独立使用。

本文gtalk server架构介绍整理自视频 Seattle Conference on Scalability: Lessons In Building Scalable Systems 而得，加入Tim少量理解和补充。

演讲者 Reza Behforooz , 是 Google Talk servers的 team leader

1. Google talk server基于 java 平台实现，见视频23:50问答。

2. 关于系统设计与实现难点
和任何xmpp server一样，难点是处理Presence峰值流量，而不是用户数或并发用户数
峰值 total QPS > 10万
Presence = ConnectedUsers * BuddyListSize * OnlineStateChanges
系统在一夜之间通过 gmail/orkut用户突然增大，没有一个缓慢适应的过程。

3. 关于压力测试
1) 实验室式的压力测试只是一个开端;
2) 在正式上线之前，通过真实环境的试验平台，选取大约10%的用户，来做实际环境测试，在不显示UI的前提下，将google talk代码嵌入用户页面
这样在gtalk正式上线时候，所有功能已经是比较有把握的。

4. 数据和应用服务分布到多服务器
可以动态扩容，升级和更新，无须停机
但是 google talk server服务并不考虑跨 IDC, 因为作者认为在海量流量处理下，跨数据中心不利于系统处理。
分布包括数据分布，可参看以前文章MySQL 分表分数据库服务器的一种方案HSCALE, 基于MySQL proxy 另外还有请求和逻辑服务分布

5. abstraction, 抽象与分离
不将问题带入别的系统, gmail/gtalk单独出问题不影响另外一方使用。
gmail, orkut 与 google talk server 完全没有关系
如 gmail 无须关心 google talk 有哪些服务器，有多少，在哪里，怎样分布。
不同系统之间通过 gateway(logic name) 来访问, gateway 再映射到物理的服务器。

6. 避免客户端或服务器自动重连造成DoS问题，服务器瞬时访问过大而瘫痪

7. 在服务器增加profile/monitor机制，包括配置文件，资源状况, 日志, 可以做离线分析。
这样某个服务器发现问题，首先去看 monitor console
profile/monitor不是侵入式的，就是不是通过在服务器程序嵌入代码实现。

8. Google Talk Server其他一些开发经验
程序可以先前和向后兼容(协议/API兼容)，可以逐个或批量升级，新老系统可以共存
有良好的平台来试验新功能，就是上文中提到可以用gmail来试验gtalk，这个是很多公司不具备的
开发人员可以直接访问真实生产环境(production environment)，观察系统实际运行情况来调优和改进。

最近看到了不少Facebook chat技术架构的介绍，如 InfoQ 和 Inside Facebook。总结如下，公开的资料以概念为主，没有什么新的或特殊的亮点。

Presence notification

• Facebook Chat 与一般的IM系统一样。最复杂之处不是消息发送，而是在线状态通知。而facebook为了提高用户体验，考虑增加的功能更增大这些处理，比如好友可以看到某个用户“idle-for-10-minutes”，用户即使不操作也有巨大负荷。
• 处理量是：O(平均好友数 * 在线用户峰值 * 变化频率) msg /sec ?

Comet

• Facebook 网页上是一种特殊的comet, 每打开一个页面调用一个iframe, iframe直至有消息才会输出。Polling + push?
• Comet不适合用Apache来跑, 可参看comet server之java实现:asyncweb,jetty,tomcat
• Facebook使用基于epoll技术的erlang web server

Erlang

• Why Erlang? 因为erlang有天生的并发支持，不需要one thread per connection，轻量级的 "processes"
• Erlang跟其他语言交互式是选用Erlang的最大问题，Facebook为了解决这个问题，自己开发了open source的Thrift,实现远程RPC

因暂时没计划用Erlang, 借鉴的地方不多，想试下Java7 Doug Lea 的 Fork/Join

(图片来源为 blog.rd2inc.com)

Twitter的手机访问界面m.twitter.com每次访问都需要输入密码，很不方便。看了一下Twitter API发文比较简单，原理为：

Post a status update, authenticated: curl -u email:password -d status="your message here" http://twitter.com/statuses/update.xml

因此就可利用curl写个脚本，把自己用户名、密码设进去以后就不用每次登录了。

<?php
header ('Content-type: text/html; charset=utf-8');
echo '<?xml version="1.0" encoding="UTF-8"?>';

/*
POST Status update to Twitter USING PHP AND CURL
Author, Tim, iso1600 at gmail dot com
*/

// Your Twitter username/password
$my_username="username";  //change this to your login username
$my_password="password";  // change this to your login password

// What are you doing?
$status = $_POST['status'];
if ($status != '') {
 $content = "status=".urlencode($status);
 $headers  =  array( "Content-type: application/x-www-form-urlencoded" );
 
 $ch = curl_init();
 curl_setopt($ch, CURLOPT_URL, "http://twitter.com/statuses/update.xml");
 
 curl_setopt($ch, CURLOPT_RETURNTRANSFER, 1);
 curl_setopt($ch, CURLOPT_TIMEOUT, 4);
 curl_setopt($ch, CURLOPT_HTTPHEADER, $headers);
 curl_setopt($ch, CURLOPT_USERPWD, $my_username.':'.$my_password);
 curl_setopt($ch, CURLOPT_POSTFIELDS, $content);
 
 $data = curl_exec($ch);
 
 if (curl_errno($ch)) {
  print curl_error($ch);
 } else {
   curl_close($ch);
 }
}
?>
<!DOCTYPE html PUBLIC "-//WAPFORUM//DTD XHTML Mobile 1.0//EN" "http://www.wapforum.org/DTD/xhtml-mobile10.dtd">
<html xmlns="http://www.w3.org/1999/xhtml">
<head><title>Twitter mobile post</title></head><body>
<form method="post" action="twitter.php"> <input type="text" name="status"> <input type="submit" > </form>
<?php

// Get your twitter homepage.
$myt = curl_init();
curl_setopt($myt, CURLOPT_URL, "http://m.twitter.com/".$my_username);
$html = curl_exec($myt);
$p1 = strpos($html, "<body ");
$p2 = strpos($html, ">", p1) + 1;
$p3 = strpos($html, "</body>");
echo substr($html, $p2, $p3);
?>
</body></html>

写完后传到某虚拟主机测试，发现调用失败，因为大部分虚拟主机不支持curl。
那只得改用socket, php4, 按最低配置写，经过PC和手机测试都可以成功更新状态。
有PHP虚拟主机的朋友可以上传到某个隐藏路径，方便自己用，改下用户和密码就行了。

<?php
header ('Content-type: text/html; charset=utf-8');
echo '<?xml version="1.0" encoding="UTF-8"?>';
/*
POST Status update to Twitter USING PHP4 AND socket
Author, Tim, iso1600 at gmail dot com
*/

// Your Twitter username/password
$my_username="username";  //change this to your login username
$my_password="password";  // change this to your login password

// What are you doing?
$status = $_POST['status'];
if ($status != '') {
 $content = "status=".urlencode($status);
 $headers = "Authorization: Basic " . base64_encode("$my_username:$my_password");
 do_post_request("http://twitter.com/statuses/update.xml", $content, $headers); 
}
?>
<!DOCTYPE html PUBLIC "-//WAPFORUM//DTD XHTML Mobile 1.0//EN" "http://www.wapforum.org/DTD/xhtml-mobile10.dtd">
<html xmlns="http://www.w3.org/1999/xhtml">
<head><title>Twitter mobile post</title></head><body>
<form method="post" action="twitter.php"> <input type="text" name="status"> <input type="submit" > </form>
<?php

// Get your twitter homepage
$html = do_get_request("http://m.twitter.com/".$my_username);
$p1 = strpos($html, "<body ");
$p2 = strpos($html, ">", p1) + 1;
$p3 = strpos($html, "</body>");
echo substr($html, $p2, $p3);

function do_post_request($url, $data, $optional_headers = null) {
 $start = strpos($url,'//')+2;
 $end = strpos($url,'/',$start);
 $host = substr($url, $start, $end-$start);
 $domain = substr($url,$end);
 $fp = pfsockopen($host, 80);
 if(!$fp) return null;
 fputs ($fp,"POST $domain HTTP/1.1\n");
 fputs ($fp,"Host: $host\n");
 
 if ($optional_headers) {
  fputs($fp, $optional_headers);
 }
 fputs ($fp,"Content-type: application/x-www-form-urlencoded\n");
 fputs ($fp,"Content-length: ".strlen($data)."\n\n");
 fputs ($fp,"$data\n\n");

 $response = "";
 while(!feof($fp)) {
  $response .= fgets($fp, 1024);
 }
 fclose ($fp);
 return $response;
}

function do_get_request($url) {
    $start = strpos($url,'//')+2;
    $end = strpos($url,'/',$start);
    $host = substr($url, $start, $end-$start);
    $domain = substr($url,$end);
    $fp = pfsockopen($host, 80);
    if(!$fp) return null;
    fputs ($fp,"GET $domain HTTP/1.0\n");
    fputs ($fp,"Host: $host\n\n");
        
    $response = ""; 
    while(!feof($fp)) {
        $response .= fgets($fp, 1024);
    }   
    fclose ($fp);
 $start = strpos($response, "\r\n\r\n");
    return substr($response, $start);
}
?>
</body></html>