python - 如何将redis键添加到另一个实例

Question

我正在尝试将现有 key 从 Redis 实例 (A) 移动到另一个实例 (B) 而不会丢失 B 的 key 。 有什么工具可以使用还是我应该自己写代码？

Answer 1

根据 Redis 官方集群教程:“......然而，目前 redis-trib 无法自动重新平衡集群，检查集群节点间键的分布，并根据需要智能地移动插槽。将添加此功能将来。”

因此您需要自己编写代码将 key 分发给 Redis 实例。如果要将一个或多个键从一个节点迁移到另一个节点，则必须先迁移哈希槽，然后再迁移实际键，否则会发生错误。

通常，您需要一个代码过程来执行这些 Ruby 代码行(redis-trib.rb 中的原型(prototype)):

将槽从源节点迁移到目标节点的代码

target.r.cluster("setslot",slot,"importing",source.info[:name])
source.r.cluster("setslot",slot,"migrating",target.info[:name])

迁移实际 key 的代码

source.r.client.call(["migrate",target.info[:host],target.info[:port],$keys_of_source[i],0,15000]).

您可以使用现有代码从此处进行迁移:https://github.com/projecteru/redis-trib.py

您可以使用这些代码行:

def migrate_slots_stable(src_host, src_port, dst_host, dst_port, slots):

    if src_host == dst_host and src_port == dst_port:
        raise ValueError('The source and the target node are the same.  A good reason is that the key is stored already on target node!')
    with Connection( src_host, src_port ) as t:
        nodes, myself = _list_masters( t, src_host)
    slots = set( slots )
    print('\x1b[6;30;42m' + '*****************************************MIGRATION PROCEDURE******************************************' + '\x1b[0m')
    print('\x1b[6;30;42m' + '******************************************************************************************************' + '\x1b[0m')
    print '* Migrating slots: ', slots , "* \n"
    for n in nodes:
        if n.host == dst_host and n.port == dst_port:
            return _migr_slots_stable( myself, n, slots, nodes )
    #raise ValueError( 'Two nodes are not in the same cluster' )


def _migr_slots_stable(source_node, target_node, slots, nodes):
    key_count = 0
    for slot in slots:
        key_count += _migr_one_slot_stable(source_node, target_node, slot, nodes)
        _migr_one_slot_stable( source_node, target_node, slot, nodes )
        #print "* keys_count that exists in these slot(s): ", key_count,' *'
        #print "\n"
        #print '* Migrated succesfully:',len(slots),'slot(s) and ',key_count,'key(s) from SOURCE NODE --> node_id:',source_node.node_id,'host:',source_node.host,'port:',source_node.port,'to TARGET NODE --> node_id:', target_node.node_id,'host:',target_node.host,'port:',target_node.port,' *'
    #print('\x1b[6;30;42m' + '******************************************************************************************************' + '\x1b[0m')
    print "\n"


def _migr_one_slot_stable(source_node, target_node, slot, nodes):
    def expect_exec_ok(m, conn, slot):
        if m.lower() != 'ok':
            conn.raise_('\n'.join([
            'Error while moving slot [ %d ] between' % slot,
            'Source node - %s:%d' % (source_node.host, source_node.port),
            'Target node - %s:%d' % (target_node.host, target_node.port),
            'Got %s' % m]))

    # Set the new node as the owner of the slot in all the known nodes
    # Bind the hash slot to a different node
    @retry(stop_max_attempt_number=16, wait_fixed=100)
    def setslot_stable(conn, slot, node_id):
        m = conn.execute('cluster', 'setslot', slot, 'node', node_id)
        expect_exec_ok(m, conn, slot)

    source_conn = source_node.get_conn()
    target_conn = target_node.get_conn()

    # Same in Ruby with these lines of code
    # the target node importing the slot and the source node migrating it
    try:
        expect_exec_ok(
        target_conn.execute('cluster', 'setslot', slot, 'importing',
                            source_node.node_id),
        target_conn, slot)
    except hiredis.ReplyError, e:
        if 'already the owner of' not in e.message:
        target_conn.raise_(e.message)

    try:
        expect_exec_ok(
        source_conn.execute('cluster', 'setslot', slot, 'migrating',
                            target_node.node_id),
        source_conn, slot)
    except hiredis.ReplyError, e:
        if 'not the owner of' not in e.message:
        source_conn.raise_(e.message)


    keys = _migr_keys_stable(source_conn, target_node.host, target_node.port, slot)
    setslot_stable(source_conn, slot, target_node.node_id)
    for node in nodes:
        setslot_stable(node.get_conn(), slot, target_node.node_id)
    return keys

def _migr_keys_stable(src_conn, target_host, target_port, slot):
    key_count = 0
    while True:
        keys = src_conn.execute('cluster', 'getkeysinslot', slot, 10)
        if len(keys) == 0:
            return key_count
        key_count += len(keys)
        src_conn.execute_bulk(
        # Same in Ruby with this line of code:    source.r.client.call(["migrate",target.info[:host],target.info[:port],$keys_of_source[i],0,15000])
        [['migrate', target_host, target_port, k, 0, 30000] for k in keys])

触发的第一个函数是def migrate_slots_stable，参数包括源主机、源端口、目标主机、目标端口和一组要传输的哈希槽。