需要一个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>

前面简单测试了memcacheq一个国内开发的message queue之性能研究，所以想了解下目前流行的Apache ActiveMQ在相同环境的表现，先声明这个比较没有广泛的借鉴意义，因为ActiveMQ是一全面的产品，因为本人只用到部分单一功能，所以做以下测试，测试环境和上文相同，简单说明如下

• 版本：ActiveMQ 5.1.0, Java 6
• ActiveMQ使用默认配置,
• producer 和 consumer 100万条信息，每条100字节
• 使用默认Examples下面ProducerTool.java & Consumer.java 加入几行统计时间速度代码，使用单线程。
• ActiveMQ支持 async publisher (non-persistent) 和 sync publisher, 区别比较大，persistent msg client会阻塞到服务器存入成功并返回结果才完成。
• 存储使用默认的 AMQ Message Store
• 修改了 conf/activemq/activemq.xml中内存的配置，如下

     <destinationPolicy>
            <policyMap>
                <policyEntries>
                    <policyEntry queue="&gt;" memorylimit="1500mb">
                    <policyEntry topic="&gt;" memorylimit="5mb">
                        <dispatchPolicy>
                            <strictOrderDispatchPolicy>
                        </strictOrderDispatchPolicy>
                        <subscriptionRecoveryPolicy>
                            <lastImageSubscriptionRecoveryPolicy>
                        </lastImageSubscriptionRecoveryPolicy>
                    </subscriptionRecoveryPolicy>
                </dispatchPolicy>
            </policyEntry>
        </policyEntry>
</policyEntries></policyMap></destinationPolicy>

测试结果：
1. 单独 producer
10,027 async non-persistent msgs / sec
4,462 persistent msgs / sec，单位下同

2. 单独 consumer, 使用上步的数据
8,195
848

3. 同时producer 和 consumer, 当producer到10万条的时候开始 consumer
producer avg:
5,593
960
consumer avg:
4,462
1,002

结果分析：

• 同时producer&consumer偶尔有卡住3~5s的现象(可能硬盘慢造成？)，但会自动恢复
• 使用 async non-persistent 方式速度极快，对于偶尔丢失少量数据不敏感的应用极为适合。
• 使用 persistent 和 memcacheq 的速度区别不大。
• ActiveMQ 可以 tuning，因为相关配置复杂，没有进一步去研究。

XMPP中的presence适合用Message Queue的方式来实现，所以一直对快速的message queue实现比较关心。刚好在 memcachedb 上看到了一个 memcacheq 的项目，从字面上看就是一个mq的实现。因为它也采用memcache的协议，所以将以前做的memcachedb的性能测试程序稍微修改看了下结果，环境和前面一样，测试数据不承担任何责任，仅供参考：）

1. 仅写入队列, 100万条
Thread: 2
Time elpased: 427s
Avg: 2,349 puts / sec
bdb data dir size: 945M

2. 仅队列读出, 100万条，使用上一步构建的队列数据。
Thread: 2
Time elpased: 569s
Avg: 1,763 gets / sec
bdb data dir size: 1.7G (这个地方有问题，可能还没完成？)

3. 同时出入队列，这个是符合应用的真实的情况，写入10万行后才开始启动读出队列程序。
Thread: 2
Time elpased: 569s
Avg: 1,049 puts / sec
Avg: 979 gets / sec
bdb data dir size: 1.6G

测试总结：

• 测试3中入队去除刚开始单独放10万条记录的速度，实际上出队和入队的速度是一样的，即同时操作平均速度在每秒1,000条(100 byte message)。
• 速度趋于相同可参考我以前的分析 memcachedb单线程访问bdb的阻塞问题
• 测试2中数据库相关实现可能有问题，数据越跑越大。
• 因为以前用的是Apache ActiveMQ ，所以下一步打算在同环境比较一下。参看后文：ActiveMQ性能研究及与memcacheq比较

2008/5/8：看了一下memcacheq的代码，补充一下几个新看法：

• memcacheq 是基于 berkeley db queue 的 memcache 协议包装
• memcacheq 本身没有mq相关算法实现，所以上面实际测试的是单线程下berkeley db的效率。
• 空间变大的问题是由于berkeley db算法延后删除，不是问题。
• 有朋友问测试代码，其实比较简单，如下。

import java.util.concurrent.atomic.AtomicInteger;

/**
 * Test client for memcache protocol servers
 * We use java memcached client, download from http://www.whalin.com/memcached/
 * run: java MQTest [threads] [type], e.g. java MQTest 2 1
 * @author Tim
 */
public class MQTest implements Runnable {
    CacheClient cc;
    public static AtomicInteger count = new AtomicInteger();
    public static final String DATA100 =
        "TIM'S DATA AND CODE WITHOUT WARRANTY OF ANY KIND.#################" +
        "##################################";
    
    // type: 0 - get, 1 - put
    private static int testType = 1;
    public static void main(String[] args) {
        int thread = 1;
        
        if (args.length > 1) {
            thread = Integer.parseInt(args[0]);
            testType = Integer.parseInt(args[1]);
        }
        
        System.out.println("Thread: " + thread);
        System.out.println("Test Type: " + (testType == 0 ? "get queue" : "put queue"));
        
        long time1 = System.currentTimeMillis();
        for (int i = 0; i < thread; i++) {
            MQTest test = new MQTest();
            Thread t = new Thread(test);
            t.start();
        }
        
        printStat(time1);
    }

    public MQTest() {
        cc = new CacheClient();
        cc.setServerPort("server:11211");
    }

    public void run() {
        while (true) {
            if (testType != 0)
                set();
            else
                get();
        }
    }

    public void get() {
        cc.get("q");
        count.incrementAndGet();
    }

    public void set() {
        cc.set("q", DATA100);
        count.incrementAndGet();
    }
    
    private static void printStat(long time1) {
        while (true) {
            long time2 = System.currentTimeMillis();
            if (time2 == time1)
                continue;
            int cnt = count.get();
            
            System.out.println("---------------------------");
            System.out.println("Total: " + cnt);

            System.out.print("Time elapsed: ");
            System.out.println((time2 - time1) / 1000);
            
            System.out.print("AVG: ");
            System.out.println(cnt * 1000l / (time2 - time1));
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

网上看到的文章Load-balancing Tomcat with Apache 核心内容如下：

原理：apache mod_proxy 改写 Tomcat 发出的 cookie, 判断用户属于哪个服务器。
优点：Tomcat什么也不用做。不需要share session，不需要配置cluster...
图示：

现在很多网站一说到cache就想到memcached，memcached确实是非常优秀的系统，但是在某些场合，特别在不是分布式应用的场合，或者某些数据不需要分布式的访问，那它就未必是最佳的选择。

下面比较3种cache方式，用测试结果说话

• Map，严格的说不能算cache，它适合本机访问，没有过期时间，不适合大容量，不能预计长度的数据，可能会使内存耗尽。
• ehcache，可设过期时间，当超过指定内存数量，可设置淘汰算法，可输出到磁盘，可适合本机访问，也适合用作分布式缓存，分布式缓存配置和原理稍复杂，没有memcached直观，本测试未使用ehHcache分布式支持。
• Memcached, 适合分布式缓存，可设过期时间

1. 环境
OS: Linux, Ubuntu 7.04 64-bit
Memory: 4G
CPU: Intel(R) Pentium(R) D CPU 2.66GHz
SCSI DISK, ext3 file system

libevent 1.3e
Memcached 1.2.4
Java 1.6.0

2. 测试方法
Key: 数字,1~100万
数据：100字节字符串

// put 100 char
    Element e = new Element(String.valueOf(n), "blah.....blah... 100 chars...");
    cache.put(e);

Memcache的设置方法参看memcachedb的性能测试

3. ehcache 设置

<cache name="cache1"
    maxElementsInMemory="1000000"
    eternal="true"
    overflowToDisk="false"
    timeToIdleSeconds="36000"
    timeToLiveSeconds="36000"
    diskPersistent="false"
    diskExpiryThreadIntervalSeconds="120"
    memoryStoreEvictionPolicy="LRU"
    />

4. 测试结果

类型	测试数据	平均速度(次/秒)	最大速度(次/秒)	占用内存	Thread(s)
HashMap.put*	1千万	146,268	254,262	1344.18MB	1*
Hashtable.put	1千万	128,752	253,911	1572.52MB	4
EHCache.put	1千万	118,399	381,601	1245.08MB	4
Memcached.put	1百万	16,515	19,942	118.20MB	4

* HashMap 不是线程安全，多线程测试无意义。

* 未比较GET测试结果，由于GET测试先要模拟一定数据，用空表去测试GET结果可能无意义。(但是GET比较可能更重要，有时间补上)

本文地址为：http://hi.baidu.com/jabber

参考资源：NP博士的文章PHP cache的比较 《大型》系列（三）——Cache & Buffer

补充：

写完几天之后无意在网上看到这两篇文章：一正一反，

正方：Comparing Memcached and Ehcache Performance说ehcache要快50～100倍

反方：Unfair Benchmarks of Ehcache vs Memcached，貌似国内访问不到，里面意思是说要用memcached的getmulti方式测试比较才公平。另外担心ehcache中LRU算法GC不能回收内存。

• 本文讨论的是sina的memcachedb,不是memcached
• Memcachedb基于libevent实现，也就是说单线程的。理论上讲所有的IO操作都需要通过信号来处理，先注册信号，系统通知可以读写了再调用具体的读写命令。这主要针对网络，严格意义上来说本地的读写也要这样处理。否则本地阻塞会引起整个服务器阻塞。
• Memcachedb基于实用的原因，读写数据库未作信号处理，但问题也不大：
  • 使用嵌入的berkeley db，仅本程序访问。
  • 直接调用berkeley db的API操作数据库，相当于直接操作文件的性能。
  • 由于不是标准的IO调用，而是berkeley db API，要通过信号来控制也很复杂。
  • 但如果启用berkeley db的replication特性，数据库阻塞的机会可能会增大。

为什么考虑这个呢，因为今天考虑在libevent里面调用mysql，情况就复杂一点。

• 调用的MySQL client C API, 通过socket同MySQL沟通，网络可能会阻塞。
• MySQL同时服务多个应用，可能有时候会繁忙响应很慢。
• MySQL C API无法设置event

所以感觉不能用memcachedb的方法。目前想到的办法就是把所有的MySQL操作通过异步去完成，比如把操作放入队列，用另外一个程序或线程去处理。但是这只适合插入和修改，如果是查询需要即时等待返回结果就不适合，很麻烦。

本文分类：高性能服务器

附：memcachedb的存取数据库代码

void complete_nread(conn * c) {
    item *it = &(c->item);
    int comm = c->item_comm;
    int ret;
    stats.set_cmds++;
    while (1) {
        if (strncmp(ITEM_data(it) + it->nbytes - 2, "\r\n", 2) != 0) {
            out_string(c, "CLIENT_ERROR bad data chunk");
            break;
        }
        cleanup_dbt();
        if (comm == NREAD_ADD || comm == NREAD_REPLACE) {
            dbkey.data = ITEM_key(it);
            dbkey.size = strlen(ITEM_key(it));
            if ((ret = dbp->get(dbp, NULL, &dbkey, &dbdata, 0)) == 0) {
                if (comm == NREAD_ADD) {
                    out_string(c, "NOT_STORED");
                    break;
                }
            } else if (comm == NREAD_REPLACE) {
                out_string(c, "NOT_STORED");
                break;
            }
        }
        cleanup_dbt();
        dbkey.data = ITEM_key(it);
        dbkey.size = strlen(ITEM_key(it));
        dbdata.data = ITEM_data(it);
        dbdata.size = it->nbytes;
        // 这里直接调用数据库存取
        if ((ret = dbp->put(dbp, NULL, &dbkey, &dbdata, 0)) == 0) {
            /* some future code? */
            out_string(c, "STORED");
        } else {
            out_string(c, "NOT_STORED");
        }
        break;
    }
    return;
}