您如何为 node.js 实施 AWS Elasticache 自动发现

Question

我是一个 node 菜鸟，并试图了解如何在 node.js 应用程序中实现 auto discovery。我将使用cluster module 并希望每个工作进程都保持最新（并永久连接到）elasticache 节点。

由于没有共享内存的概念（如 PHP APC），您是否必须让代码在每个工作人员中运行，每 X 秒唤醒一次并以某种方式更新 IP 列表并重新连接 memcache 客户端？

今天人们如何解决这个问题？示例代码将不胜感激。

Answer 1

请注意，目前，自动发现仅适用于运行 memcached 引擎的缓存集群。

对于 1.4.14 或更高版本的缓存引擎，您需要创建一个到缓存集群配置端点（或任何缓存节点端点）的 TCP/IP 套接字并发送以下命令：

config get cluster

使用 Node.js，您可以使用 net.Socket class 来实现。

回复由两行组成：

• 配置信息的版本号。每次从缓存集群中添加或删除节点时，版本号都会增加一。

• 缓存节点列表。列表中的每个节点都由一个主机名|IP地址|端口组表示，每个节点用空格分隔。

回车和换行符 (CR + LF) 出现在每行的末尾。

您可以在此处找到有关如何add Auto Discovery to your client library 的更详尽说明。

使用集群模块，您需要在每个进程（即子进程）中存储相同的信息，我会为每个子进程使用“setInterval”来定期检查（例如每 60 秒）节点列表并重新连接 仅当列表发生变化时（这种情况不应该经常发生）。

您可以选择仅更新主服务器上的列表，并使用“worker.send”更新工作人员。这可以使在单个服务器中运行的所有进程更加同步，但在多服务器架构中无济于事，因此使用一致哈希非常重要，以便能够更改节点列表并释放存储在 memcached 集群中的“最小”数量的键。

我会使用一个全局变量来存储这种配置。

三思而后行，您可以使用 AWS SDK for Node.js 来获取 ElastiCache 节点列表（这也适用于 Redis 引擎）。

在这种情况下，代码将类似于：

var util = require('util'),
    AWS = require('aws-sdk'),
    Memcached = require('memcached');

global.AWS_REGION = 'eu-west-1'; // Just as a sample I'm using the EU West region                                                                                     
global.CACHE_CLUSTER_ID = 'test';
global.CACHE_ENDPOINTS = [];
global.MEMCACHED = null;

function init() {

    AWS.config.update({
        region: global.AWS_REGION
    });
    elasticache = new AWS.ElastiCache();

    function getElastiCacheEndpoints() {

        function sameEndpoints(list1, list2) {
            if (list1.length != list2.length)
                return false;
            return list1.every(
                function(e) {
                    return list2.indexOf(e) > -1;
                });
        }

        function logElastiCacheEndpoints() {
            global.CACHE_ENDPOINTS.forEach(
                function(e) {
                    util.log('Memcached Endpoint: ' + e);
                });
        }

        elasticache.describeCacheClusters({
                CacheClusterId: global.CACHE_CLUSTER_ID,
                ShowCacheNodeInfo: true
            },
            function(err, data) {
                if (!err) {
                    util.log('Describe Cache Cluster Id:' + global.CACHE_CLUSTER_ID);
                    if (data.CacheClusters[0].CacheClusterStatus == 'available') {
                        var endpoints = [];

                        data.CacheClusters[0].CacheNodes.forEach(
                            function(n) {
                                var e = n.Endpoint.Address + ':' + n.Endpoint.Port;
                                endpoints.push(e);
                            });
                        if (!sameEndpoints(endpoints, global.CACHE_ENDPOINTS)) {
                            util.log('Memached Endpoints changed');
                            global.CACHE_ENDPOINTS = endpoints;
                            if (global.MEMCACHED)
                                global.MEMCACHED.end();
                            global.MEMCACHED = new Memcached(global.CACHE_ENDPOINTS);
                            process.nextTick(logElastiCacheEndpoints);
                            setInterval(getElastiCacheEndpoints, 60000); // From now on, update every 60 seconds                        
                        }
                    } else {
                        setTimeout(getElastiCacheEndpoints, 10000); // Try again after 10 seconds until 'available'                     
                    }
                } else {
                    util.log('Error describing Cache Cluster:' + err);
                }
            });
    }

    getElastiCacheEndpoints();

}

init();