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

Question

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

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

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

Answer 1

Note that at this time, Auto Discovery is only available for cache clusters running the memcached engine.

For Cache Engine Version 1.4.14 or Higher you need to create a TCP/IP socket to the Cache Cluster Configuration Endpoint (or any Cache Node Endpoint) and send this command:

config get cluster

With Node.js you can use the net.Socket class to to that.

The reply consists of two lines:

• The version number of the configuration information. Each time a node is added or removed from the cache cluster, the version number increases by one.

• A list of cache nodes. Each node in the list is represented by a hostname|ip-address|port group, and each node is delimited by a space.

A carriage return and a linefeed character (CR + LF) appears at the end of each line.

Here you can find a more thorough description of how to add Auto Discovery to your client library.

Using the cluster module you need to store the same information in each process (i.e. child) and I would use "setInterval" per child to periodically check (e.g. every 60 seconds) the list of nodes and re-connect only if the list has changed (this should not happen very often).

You can optionally update the list on the master only and use "worker.send" to update the workers. This could keep all the processes running in a single server more in sync, but it would not help in a multi server architecture, so it is very important to use consistent hashing in order to be able to change the list of nodes and loose the "minimum" amount of keys stored in the memcached cluster.

I would use a global variable to store this kind of configuration.

Thinking twice you can use the AWS SDK for Node.js to get the list of ElastiCache Nodes (and that works for the Redis engine as well).

In that case the code would be something like:

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();