实现 Redis 集群

    A
   / \
  D   B
   \ /
    C

package consistenthash
 
type NodeMap struct {
	hashFunc  func(data []byte) uint32
	nodeHashs []int
    nodehashMap map[int]string
}

func NewNodeMap(hashFunc func(data []byte) uint32) *NodeMap {
	m := &NodeMap{
		hashFunc:    hashFunc,
		nodehashMap: make(map[int]string),
	}
	if m.hashFunc == nil {
		m.hashFunc = crc32.ChecksumIEEE
	}
	return m
}

func (m *NodeMap) IsEmpty() bool {
	return len(m.nodehashMap) == 0
}

func (m *NodeMap) AddNodes(nodes ...string) {
	for _, node := range nodes {
		if node == "" {
			continue
		}
		hash := int(m.hashFunc([]byte(node)))
		m.nodeHashs = append(m.nodeHashs, hash)
		m.nodehashMap[hash] = node
	}
	sort.Ints(m.nodeHashs)
}

sort.Search(n, f func(i int) bool) int

// PickNode picks a node based on the key, returning the node that is closest to the hash of the key
func (m *NodeMap) PickNode(key string) string {
	if m.IsEmpty() {
		return ""
	}
 
	hash := int(m.hashFunc([]byte(key)))
	index := sort.Search(len(m.nodeHashs), func(i int) bool {
		return m.nodeHashs[i] >= hash
	})
	// If the hash is greater than all node hashes, Int.Search returns len(m.nodeHashs)
	// So we need to wrap around to the first node
	if index == len(m.nodeHashs) {
		index = 0
	}
	return m.nodehashMap[m.nodeHashs[index]]
}

------------------------------
| cluster_database           |
|----------------------------|
| standalone_database        |
------------------------------

package database
 
import (
	"redigo/aof"
	"redigo/config"
	"redigo/interface/resp"
	"redigo/lib/logger"
	"redigo/resp/reply"
	"strconv"
	"strings"
)
 
type StandaloneDatabase struct {
	dbSet      []*DB
	aofHandler *aof.AofHandler
}
 
// NewStandaloneDatabase creates a new StandaloneDatabase instance
func NewStandaloneDatabase() *StandaloneDatabase {
	database := &StandaloneDatabase{}
	if config.Properties.Databases == 0 {
		config.Properties.Databases = 16
	}
	database.dbSet = make([]*DB, config.Properties.Databases)
	for i := range database.dbSet {
		db := MakeDB()
		db.index = i
		database.dbSet[i] = db
	}
 
	if config.Properties.AppendOnly {
		aofHandler, err := aof.NewAofHandler(database)
		if err != nil {
			panic(err)
		}
		database.aofHandler = aofHandler
		for _, db := range database.dbSet {
			// create new variable to avoid closure capturing the loop variable
			sdb := db
			sdb.addAof = func(line CmdLine) {
				database.aofHandler.AddAof(sdb.index, line)
			}
		}
	}
 
	return database
}
 
// Exec executes a command on the database
func (d *StandaloneDatabase) Exec(client resp.Connection, args [][]byte) resp.Reply {
	defer func() {
		if err := recover(); err != nil {
			logger.Error("Database Exec panic:" + err.(error).Error())
		}
	}()
	cmdName := strings.ToLower(string(args[0]))
	if cmdName == "select" {
		if len(args) != 2 {
			return reply.MakeArgNumErrReply("select")
		}
		return execSelect(client, d, args[1:])
	}
	// Get the current database index from the client connection
	db := d.dbSet[client.GetDBIndex()]
	return db.Exec(client, args)
}
 
func (d *StandaloneDatabase) AfterClientClose(c resp.Connection) {
 
}
 
func (d *StandaloneDatabase) Close() {
 
}
 
// execSelect sets the current database for the client connection.
// select x
func execSelect(c resp.Connection, database *StandaloneDatabase, args [][]byte) resp.Reply {
	dbIndex, err := strconv.Atoi(string(args[0]))
	if err != nil {
		return reply.MakeStandardErrorReply("ERR invalid DB index")
	}
	if dbIndex < 0 || dbIndex >= len(database.dbSet) {
		return reply.MakeStandardErrorReply("ERR DB index out of range")
	}
	c.SelectDB(dbIndex)
	return reply.MakeOKReply()
}

func MakeHandler() *RespHandler {
    // 修改这里
	db := database.NewStandaloneDatabase()
	return &RespHandler{
		db: db,
	}
}

type Client struct {
	conn        net.Conn
	pendingReqs chan *request // wait to send
	waitingReqs chan *request // waiting response
	ticker      *time.Ticker
	addr        string
	working     *sync.WaitGroup // its counter presents unfinished requests(pending and waiting)
}

type request struct {
	id        uint64
	args      [][]byte
	reply     resp.Reply
	heartbeat bool
	waiting   *wait.Wait
	err       error
}

func MakeClient(addr string) (*Client, error) {
	conn, err := net.Dial("tcp", addr)
	if err != nil {
		return nil, err
	}
	return &Client{
		addr:        addr,
		conn:        conn,
		pendingReqs: make(chan *request, 256),
		waitingReqs: make(chan *request, 256),
		working:     &sync.WaitGroup{},
	}, nil
}

func (client *Client) Start() {
	client.ticker = time.NewTicker(10 * time.Second)
	go client.handleWrite()
	go func() {
		err := client.handleRead()
		if err != nil {
			logger.Error(err)
		}
	}()
	go client.heartbeat()
}

// Send sends a request to redis server
func (client *Client) Send(args [][]byte) resp.Reply {
	request := &request{
		args:      args,
		heartbeat: false,
		waiting:   &wait.Wait{},
	}
	request.waiting.Add(1)
	client.working.Add(1)
	defer client.working.Done()
	client.pendingReqs <- request
	timeout := request.waiting.WaitWithTimeout(maxWait)
	if timeout {
		return reply.MakeStandardErrorReply("server time out")
	}
	if request.err != nil {
		return reply.MakeStandardErrorReply("request failed")
	}
	return request.reply
}

func (client *Client) handleWrite() {
	for req := range client.pendingReqs {
		client.doRequest(req)
	}
}

func (client *Client) doRequest(req *request) {
	if req == nil || len(req.args) == 0 {
		return
	}
	re := reply.MakeMultiBulkReply(req.args)
	bytes := re.ToBytes()
	_, err := client.conn.Write(bytes)
	i := 0
	for err != nil && i < 3 {
		err = client.handleConnectionError(err)
		if err == nil {
			_, err = client.conn.Write(bytes)
		}
		i++
	}
	if err == nil {
		client.waitingReqs <- req
	} else {
		req.err = err
		req.waiting.Done()
	}
}

func (client *Client) handleRead() error {
	ch := parser.ParseStream(client.conn)
	for payload := range ch {
		if payload.Err != nil {
			client.finishRequest(reply.MakeStandardErrorReply(payload.Err.Error()))
			continue
		}
		client.finishRequest(payload.Data)
	}
	return nil
}

func (client *Client) finishRequest(reply resp.Reply) {
	defer func() {
		if err := recover(); err != nil {
			debug.PrintStack()
			logger.Error(err)
		}
	}()
	request := <-client.waitingReqs
	if request == nil {
		return
	}
	request.reply = reply
	if request.waiting != nil {
		request.waiting.Done()
	}
}

func (client *Client) heartbeat() {
	for range client.ticker.C {
		client.doHeartbeat()
	}
}
 
func (client *Client) doHeartbeat() {
	request := &request{
		args:      [][]byte{[]byte("PING")},
		heartbeat: true,
		waiting:   &wait.Wait{},
	}
	request.waiting.Add(1)
	client.working.Add(1)
	defer client.working.Done()
	client.pendingReqs <- request
	request.waiting.WaitWithTimeout(3 * time.Second)
}

func (client *Client) handleConnectionError(err error) error {
	err1 := client.conn.Close()
	if err1 != nil {
		if opErr, ok := err1.(*net.OpError); ok {
			if opErr.Err.Error() != "use of closed network connection" {
				return err1
			}
		} else {
			return err1
		}
	}
	conn, err1 := net.Dial("tcp", client.addr)
	if err1 != nil {
		logger.Error(err1)
		return err1
	}
	client.conn = conn
	go func() {
		_ = client.handleRead()
	}()
	return nil
}

func (client *Client) Close() {
	client.ticker.Stop()
	close(client.pendingReqs)
	client.working.Wait()
	_ = client.conn.Close()
	close(client.waitingReqs)
}

type Database interface {
	Exec(client resp.Connection, args [][]byte) resp.Reply
	AfterClientClose(c resp.Connection)
	Close()
}

type ClusterDatabase struct {
}
 
// MakeClusterDatabase creates a new ClusterDatabase instance
func MakeClusterDatabase() *ClusterDatabase {
	return &ClusterDatabase{}
}
 
// Exec executes a command on the cluster database
func (c *ClusterDatabase) Exec(client resp.Connection, args [][]byte) resp.Reply {
	return nil
}
 
// Close closes the cluster database
func (c *ClusterDatabase) Close() {
}
 
// AfterClientClose is called after a client closes
func (c *ClusterDatabase) AfterClientClose(client resp.Connection) {
}

go get github.com/jolestar/go-commons-pool/v2

type ClusterDatabase struct {
	self       string                      // self node id
	nodes      []string                    // cluster nodes
	peerPicker *consistenthash.NodeMap     // consistent hash ring
	peerConn   map[string]*pool.ObjectPool // connection pool for each node
	db         database.Database           // database instance
}

// ObjectPool is a generic object pool
type ObjectPool struct {
	AbandonedConfig                  *AbandonedConfig
	Config                           *ObjectPoolConfig
	closed                           bool
	closeLock                        sync.Mutex
	evictionLock                     sync.Mutex
	idleObjects                      *collections.LinkedBlockingDeque
	allObjects                       *collections.SyncIdentityMap
	factory                          PooledObjectFactory
	createCount                      concurrent.AtomicInteger
	destroyedByEvictorCount          concurrent.AtomicInteger
	destroyedCount                   concurrent.AtomicInteger
	destroyedByBorrowValidationCount concurrent.AtomicInteger
	evictor                          *time.Ticker
	evictorStopChan                  chan struct{}
	evictorStopWG                    sync.WaitGroup
	evictionIterator                 collections.Iterator
}
 
// NewObjectPool return new ObjectPool, init with PooledObjectFactory and ObjectPoolConfig
func NewObjectPool(ctx context.Context, factory PooledObjectFactory, config *ObjectPoolConfig) *ObjectPool {
	return NewObjectPoolWithAbandonedConfig(ctx, factory, config, nil)
}
 
// NewObjectPoolWithDefaultConfig return new ObjectPool init with PooledObjectFactory and default config
func NewObjectPoolWithDefaultConfig(ctx context.Context, factory PooledObjectFactory) *ObjectPool {
	return NewObjectPool(ctx, factory, NewDefaultPoolConfig())
}
 
// NewObjectPoolWithAbandonedConfig return new ObjectPool init with PooledObjectFactory, ObjectPoolConfig, and AbandonedConfig
func NewObjectPoolWithAbandonedConfig(ctx context.Context, factory PooledObjectFactory, config *ObjectPoolConfig, abandonedConfig *AbandonedConfig) *ObjectPool {
	pool := ObjectPool{factory: factory, Config: config,
		idleObjects:             collections.NewDeque(math.MaxInt32),
		allObjects:              collections.NewSyncMap(),
		createCount:             concurrent.AtomicInteger(0),
		destroyedByEvictorCount: concurrent.AtomicInteger(0),
		destroyedCount:          concurrent.AtomicInteger(0),
		AbandonedConfig:         abandonedConfig}
	pool.StartEvictor()
	return &pool
}

type PooledObjectFactory interface {
 
	/**
	 * Create a pointer to an instance that can be served by the
	 * pool and wrap it in a PooledObject to be managed by the pool.
	 *
	 * return error if there is a problem creating a new instance,
	 *    this will be propagated to the code requesting an object.
	 */
	MakeObject(ctx context.Context) (*PooledObject, error)
 
	/**
	 * Destroys an instance no longer needed by the pool.
	 */
	DestroyObject(ctx context.Context, object *PooledObject) error
 
	/**
	 * Ensures that the instance is safe to be returned by the pool.
	 *
	 * return false if object is not valid and should
	 *         be dropped from the pool, true otherwise.
	 */
	ValidateObject(ctx context.Context, object *PooledObject) bool
 
	/**
	 * Reinitialize an instance to be returned by the pool.
	 *
	 * return error if there is a problem activating object,
	 *    this error may be swallowed by the pool.
	 */
	ActivateObject(ctx context.Context, object *PooledObject) error
 
	/**
	 * Uninitialize an instance to be returned to the idle object pool.
	 *
	 * return error if there is a problem passivating obj,
	 *    this exception may be swallowed by the pool.
	 */
	PassivateObject(ctx context.Context, object *PooledObject) error
}

type connectionFactory struct {
	Peer string // peer node id
}

// MakeObject creates a new connection object
func (f *connectionFactory) MakeObject(ctx context.Context) (*pool.PooledObject, error) {
	c, err := client.MakeClient(f.Peer)
	if err != nil {
		return nil, err
	}
	c.Start()
	return pool.NewPooledObject(c), nil
}

// DestroyObject destroys a connection object
func (f *connectionFactory) DestroyObject(ctx context.Context, pooledObject *pool.PooledObject) error {
	c, ok := pooledObject.Object.(*client.Client)
	if !ok {
		return errors.New("invalid connection type")
	}
	c.Close()
	return nil
}

// ValidateObject validates a connection object
// We don't need it, just return true
func (f *connectionFactory) ValidateObject(ctx context.Context, pooledObject *pool.PooledObject) bool {
	return true
}
 
// PassivateObject passivates a connection object
// We don't need it, just return nil
func (f *connectionFactory) PassivateObject(ctx context.Context, pooledObject *pool.PooledObject) error {
	return nil
}
 
// ActivateObject activates a connection object
// We don't need it, just return nil
func (f *connectionFactory) ActivateObject(ctx context.Context, pooledObject *pool.PooledObject) error {
	return nil
}

self 127:0.0.1:6379
peers 127:0.0.1:6380,127:0.0.1:6381

// MakeClusterDatabase creates a new ClusterDatabase instance
func MakeClusterDatabase() *ClusterDatabase {
    cluster := &ClusterDatabase{
        // 从配置中读取当前节点 ID
        self:       config.Properties.Self,
        // 创建一个本地的 StandaloneDatabase 实例
        db:         databaseinstance.NewStandaloneDatabase(),
        // 初始化一致性哈希环
        peerPicker: consistenthash.NewNodeMap(nil),
        // 初始化存储对等节点连接池的 map
        peerConn:   make(map[string]*pool.ObjectPool),
    }
    // 准备节点列表，包含所有 peer 和自身
    nodes := make([]string, 0, len(config.Properties.Peers)+1)
    // 添加所有 peer 节点
    nodes = append(nodes, config.Properties.Peers...)
    // 添加自身节点
    nodes = append(nodes, config.Properties.Self)
    // 将所有节点添加到一致性哈希环中
    cluster.peerPicker.AddNodes(nodes...)
    ctx := context.Background()
    // 为每个 peer 节点创建连接池
    for _, peer := range config.Properties.Peers {
        // 使用之前定义的 connectionFactory 和默认配置创建对象池
        cluster.peerConn[peer] = pool.NewObjectPoolWithDefaultConfig(ctx, &connectionFactory{
            Peer: peer,
        })
    }
    // 保存完整的节点列表
    cluster.nodes = nodes
    return cluster
}

// getPeerClient retrieves a client for the specified peer node
func (c *ClusterDatabase) getPeerClient(peer string) (*client.Client, error) {
	pool, ok := c.peerConn[peer]
	if !ok {
		return nil, errors.New("peer not found")
	}
	conn, err := pool.BorrowObject(context.Background())
	if err != nil {
		return nil, err
	}
	client, ok := conn.(*client.Client)
	if !ok {
		return nil, errors.New("invalid connection type")
	}
	return client, nil
}

// returnPeerClient returns a client to the specified peer node
func (c *ClusterDatabase) returnPeerClient(peer string, client *client.Client) {
	pool, ok := c.peerConn[peer]
	if !ok {
		return
	}
	// Return the client to the pool
	pool.ReturnObject(context.Background(), client)
}

// relay exec executes a command on the specified peer node
func (c *ClusterDatabase) relayExec(peer string, conn resp.Connection, args [][]byte) resp.Reply {
	if peer == c.self {
		return c.db.Exec(conn, args)
	}
	client, err := c.getPeerClient(peer)
	if err != nil {
		return reply.MakeStandardErrorReply(err.Error())
	}
	defer func() {
		c.returnPeerClient(peer, client)
	}()
	client.Send(utils.ToCmdLine("SELECT", strconv.Itoa(conn.GetDBIndex())))
	return client.Send(args)
}

// broadcastExec executes a command on all peer nodes
func (c *ClusterDatabase) broadcastExec(conn resp.Connection, args [][]byte) map[string]resp.Reply {
	results := make(map[string]resp.Reply)
	for _, peer := range c.nodes {
		result := c.relayExec(peer, conn, args)
		results[peer] = result
	}
	return results
}

type CmdFunc func(cluster *ClusterDatabase, conn resp.Connection, args [][]byte) resp.Reply

// defaultFunc is a default function that executes a command on the cluster database
func defaultFunc(cluster *ClusterDatabase, conn resp.Connection, args [][]byte) resp.Reply {
	key := string(args[1])
	peer := cluster.peerPicker.PickNode(key)
	return cluster.relayExec(peer, conn, args)
}

func makeRouter() map[string]CmdFunc {
	routerMap := make(map[string]CmdFunc)
	routerMap["exists"] = defaultFunc // exists key
	routerMap["type"] = defaultFunc   // type key
	routerMap["set"] = defaultFunc    // set key
	routerMap["get"] = defaultFunc    // get key
	routerMap["setnx"] = defaultFunc  // setnx key
	routerMap["getset"] = defaultFunc // getset key
 
	return routerMap
}

var routerMap = makeRouter()

// pingFunc is a function that executes a command on the cluster database
func pingFunc(cluster *ClusterDatabase, conn resp.Connection, args [][]byte) resp.Reply {
	return cluster.db.Exec(conn, args)
}

// renameFunc is a function that executes a command on the cluster database
func renameFunc(cluster *ClusterDatabase, conn resp.Connection, args [][]byte) resp.Reply {
	if len(args) != 3 {
		return reply.MakeStandardErrorReply("ERR wrong number of arguments for 'rename' command")
	}
	src := string(args[1])
	dest := string(args[2])
 
	srcPeer := cluster.peerPicker.PickNode(src)
	destPeer := cluster.peerPicker.PickNode(dest)
 
	if srcPeer != destPeer {
		return reply.MakeStandardErrorReply("ERR source and destination keys are on different nodes")
	}
 
	return cluster.relayExec(srcPeer, conn, args)
}

// flushDBFunc is a function that executes a command on the cluster database
func flushDBFunc(cluster *ClusterDatabase, conn resp.Connection, args [][]byte) resp.Reply {
	replies := cluster.broadcastExec(conn, args)
	var errReply reply.ErrorReply
	for _, r := range replies {
		if reply.IsErrReply(r) {
			errReply = r.(reply.ErrorReply)
			break
		}
	}
	if errReply == nil {
		return reply.MakeOKReply()
	}
	return reply.MakeStandardErrorReply("error: " + errReply.Error())
}

// delFunc is a function that executes a command on the cluster database
// delFunc is a function that executes a command on the cluster database
func delFunc(cluster *ClusterDatabase, c resp.Connection, args [][]byte) resp.Reply {
	// Check the number of arguments
	if len(args) < 2 {
		return reply.MakeArgNumErrReply("del")
	}
 
	// If there is only one key, route directly to the corresponding node
	if len(args) == 2 {
		key := string(args[1])
		peer := cluster.peerPicker.PickNode(key)
		// Note: The full command, including "DEL", needs to be passed
		fullArgs := make([][]byte, 2)
		fullArgs[0] = []byte("DEL")
		fullArgs[1] = args[1]
		return cluster.relayExec(peer, c, fullArgs)
	}
 
	// --- Modification starts ---
	// Handle multiple keys: group keys by node
	groupedKeys := make(map[string][][]byte) // key: peer address, value: list of keys handled by the peer
	for i := 1; i < len(args); i++ {         // Iterate over all keys to delete, starting from index 1
		key := string(args[i])
		peer := cluster.peerPicker.PickNode(key)
		if _, ok := groupedKeys[peer]; !ok {
			groupedKeys[peer] = make([][]byte, 0)
		}
		groupedKeys[peer] = append(groupedKeys[peer], args[i]) // Add the original []byte key to the list
	}
 
	// Execute delete operation for each node
	var deleted int64 = 0
	var firstErrReply reply.ErrorReply // Save the first encountered error
 
	for peer, keys := range groupedKeys {
		// Construct the DEL command for the current node: ["DEL", key1, key2, ...]
		nodeArgs := make([][]byte, len(keys)+1)
		nodeArgs[0] = []byte("DEL") // The command itself
		copy(nodeArgs[1:], keys)    // Copy the list of keys handled by this node
 
		// Send the command to the specific node
		nodeReply := cluster.relayExec(peer, c, nodeArgs)
 
		// Handle the response
		if reply.IsErrReply(nodeReply) {
			// If it is an error response, record the first error and stop processing other nodes (optional, can also choose to continue processing other nodes)
			if firstErrReply == nil {
				if errReply, ok := nodeReply.(reply.ErrorReply); ok {
					firstErrReply = errReply
				} else {
					firstErrReply = reply.MakeStandardErrorReply("unknown error from peer")
				}
			}
			// You can choose to break or continue here, depending on whether you want the entire operation to fail if one node fails
			// break // Stop and return an error if one node fails
			continue // Continue attempting to delete keys on other nodes, then summarize results or return the first error
		}
 
		// If it is an integer response, accumulate the number of deleted keys
		if intReply, ok := nodeReply.(*reply.IntReply); ok {
			deleted += intReply.Code
		} else {
			// If the response is neither the expected integer nor an error, treat it as an error
			if firstErrReply == nil {
				firstErrReply = reply.MakeStandardErrorReply("unexpected reply type from peer")
			}
			// break // Same as above
			continue // Same as above
		}
	}
 
	// If an error was encountered during processing, return the first error
	if firstErrReply != nil {
		// You can choose to return more detailed error information or just the first error
		return reply.MakeStandardErrorReply("error occurs during multi-key delete: " + firstErrReply.Error())
	}
 
	// If all nodes succeeded (or partial errors were ignored), return the total number of deleted keys
	return reply.MakeIntReply(deleted)
}

// selectFunc is a function that executes a command on the cluster database
func selectFunc(cluster *ClusterDatabase, conn resp.Connection, args [][]byte) resp.Reply {
	return cluster.db.Exec(conn, args)
}

func makeRouter() map[string]CmdFunc {
	routerMap := make(map[string]CmdFunc)
	routerMap["exists"] = defaultFunc // exists key
	routerMap["type"] = defaultFunc   // type key
	routerMap["set"] = defaultFunc    // set key
	routerMap["get"] = defaultFunc    // get key
	routerMap["setnx"] = defaultFunc  // setnx key
	routerMap["getset"] = defaultFunc // getset key
 
	routerMap["ping"] = pingFunc     // ping command
	routerMap["rename"] = renameFunc // rename key
	routerMap["renamex"] = renameFunc
	routerMap["flushdb"] = flushDBFunc // flushdb command
	routerMap["del"] = delFunc         // del key
	routerMap["select"] = selectFunc   // select database
 
	return routerMap
}

// Close closes the cluster database
func (c *ClusterDatabase) Close() {
	c.db.Close()
}
 
// AfterClientClose is called after a client closes
func (c *ClusterDatabase) AfterClientClose(client resp.Connection) {
	c.db.AfterClientClose(client)
}

// Exec executes a command on the cluster database
func (c *ClusterDatabase) Exec(client resp.Connection, args [][]byte) (result resp.Reply) {
	defer func() {
		if err := recover(); err != nil {
			logger.Error("ClusterDatabase Exec panic:" + err.(error).Error())
			result = reply.MakeUnknownReply()
		}
	}()
 
	cmdName := strings.ToLower(string(args[0]))
 
	if cmdFunc, ok := routerMap[cmdName]; ok {
		return cmdFunc(c, client, args)
	} else {
		result = reply.MakeStandardErrorReply("ERR unknown command '" + cmdName + "'")
	}
 
	return
}

// MakeHandler creates a RespHandler instance
func MakeHandler() *RespHandler {
	var db databaseface.Database
	// If self is not empty, it means this is a cluster node
	// and we need to create a cluster database
	if config.Properties.Self != "" && len(config.Properties.Peers) > 0 {
		db = cluster.MakeClusterDatabase()
	} else {
		db = database.NewStandaloneDatabase()
	}
	return &RespHandler{
		db: db,
	}
}

bind 0.0.0.0
port 6379
databases 16
appendonly yes
appendfilename appendonly.aof
self 127.0.0.1:6379
peers 127.0.0.1:6380

bind 0.0.0.0
port 6380
databases 16
appendonly yes
appendfilename appendonly.aof
self 127.0.0.1:6380
peers 127.0.0.1:6379

redis-cli -h 127.0.0.1 -p 6379
redis-cli -h 127.0.0.1 -p 6380

PING

PING

SET key1 value1

GET key1

SET key1 value1

GET key1

SET key2 value2

GET key2

DEL key1