redis

Syntax

connect(host, port)

Details

The method establishes a connection to the redis server and returns a handle for the redis connection.

Parameters

• host: A STRING scalar indicating the IP address of the redis server.
• port: An INT scalar indicating the port number of the redis server.

Examples

Assuming the redis server is listening on 127.0.0.1:6379.

conn = redis::connect("127.0.0.1",6379)

Syntax

run(conn, arg1, arg2, arg3, ...)

Details

The method executes redis commands. If the redis is password-protected, use run(conn, "AUTH", "password") to gain access. The results of the redis commands will be returned in DolphinDB type with automatic conversion.

Note: If an error occurs during execution, the connection cannot be used anymore. Release this connection and reconnect.

Parameters

• conn: The redis connection handle obtained through connect.
• arg1: A STRING scalar indicating redis commands such as SET, GET, etc.
• arg2, arg3, ...: Parameters required by the redis command.

Examples

In the following example, the run method is first used to execute the SET command, setting “abc“ to hold the string “vabc“. Then, the GET command is executed to get the value of “abc“.

conn = redis::connect("127.0.0.1",6379)
redis::run(conn, "SET", "abc", "vabc")
val = redis::run(conn, "GET", "abc")
val == "vabc"

Syntax

batchSet(conn, keys, values)

Details

The method sets the given keys to their respective values. It can be used with the subscribeTable function to facilitate the bulk data insertion into a redis database.

Note: If an error occurs during execution, the connection cannot be used anymore. Release this connection and reconnect.

Parameters

• conn: The Redis connection handle obtained through connect.
• keys: A STRING scalar or vector indicating the keys to set.
• values: A STRING scalar or vector indicating the values to set.

Examples

conn = redis::connect("127.0.0.1",6379)

redis::batchSet(conn, "k1", "v1")

keys = ["k2", "k3", "k4"]
values = ["v2", "v3", "v4"]
redis::batchSet(conn, keys, values)

Syntax

batchHashSet(conn, ids, fieldData)

Details

The method sets the specified fields to their respective values into redis hashes. It is used to batch write data to the redis database.

Note: If an error occurs during execution, the connection cannot be used anymore. Release this connection and reconnect.

Parameters

• conn: Redis connection handle obtained through connect.
• ids: A STRING array, where each element serves as a key of the Redis hash. These keys correspond to specific rows of data in the fieldData table.
• fieldData: A table where each column is of STRING type. Each column name corresponds to a field of the hash, and the value serves as the value of the Redis hash.

Examples

loadPlugin("path/PluginRedis.txt");
go

// Generate data
n=43200
instrument_id = take("HO2305-D-",n)+string(1..n)
time_stamp = timestamp(2024.02.02T09:30:01..2024.02.02T21:30:00)
exchange = take("CFFEX",n)
last_price = rand(10.0,n)
volume = rand(100000,n)
bid_price1 = rand(10.0,n)
bid_volume1 = rand(1000,n)
bid_price2 = rand(10.0,n)
bid_volume2 = rand(1000,n)
bid_price3 = rand(10.0,n)
bid_volume3 = rand(1000,n)
bid_price4 = rand(10.0,n)
bid_volume4 = rand(1000,n)
bid_price5 = rand(10.0,n)
bid_volume5 = rand(1000,n)
t = table(instrument_id, time_stamp,exchange, last_price,volume, bid_price1,bid_volume1, bid_price2,bid_volume2,bid_price3,bid_volume3,bid_price4,bid_volume4,bid_price5,bid_volume5)
conn=redis::connect("127.0.0.1",6379)

// Batch set operation
ids = exec instrument_id from t
fieldData =  select string(time_stamp) as time_stamp,string(exchange) as exchange,string(last_price) as last_price,string(volume) as volume,string(bid_price1) as bid_price1,string(bid_volume1) as bid_volume1,string(bid_price2) as bid_price2,string(bid_volume2) as bid_volume2,string(bid_price3) as bid_price3,string(bid_volume3) as bid_volume3,string(bid_price4) as bid_price4,string(bid_volume4) as bid_volume4,string(bid_price5) as bid_price5,string(bid_volume5) as bid_volume5 from t

redis::batchHashSet(conn, ids, fieldData)

Syntax

batchPush(conn, keys, values, [pushRight=true])

Details

A Redis batch push method that allows simultaneous insertion of multiple values into specified keys.

Note: While the function does not guarantee successful execution for all keys in the batch, it ensures atomic operations for individual keys—values for a specific key are either entirely added or completely skipped.

Parameters

• conn: The Redis connection handle obtained through the connect method.
• keys: A STRING vector representing the target keys.
• values: A tuple where each element is a vector of strings representing the values to be added.
• pushRight (optional): A boolean scalar. When set to true (default), the method performs RPUSH; when set to false, it performs LPUSH.

Examples

conn = redis::connect("127.0.0.1",6379)
keys = ["k1", "k2", "k3"]
values = [["v1, v2"], ["v3", "v4", "v5"], ["v6"]]
redis::batchPush(conn, keys, values)

Syntax

redis::batchGet(conn, key)

Details

The method queries the specified keys in batches and returns a STRING vector.

Parameters

• conn: The Redis connection handle obtained through the redis::connect method.
• key: A STRING vector indicating the target keys to query.

Examples

conn = redis::connect("127.0.0.1",6379)
key = ["k2", "k3", "k4"]
redis::batchGet(conn, key)

Syntax

release(conn)

Details

The method closes the specified connection to the redis server.

Parameters

• conn: Redis connection handle obtained through connect.

Examples

conn = redis::connect("127.0.0.1",6379)
redis::release(conn)

Syntax

releaseAll()

Details

The method closes all connections to the redis server.

Examples

conn = redis::releaseAll()

Syntax

getHandle(token)

Details

The method returns a specific redis handle.

Parameters

• token: A unique identifier for a Redis connection, obtained from the first column of the table returned by getHandleStatus.

Examples

handle = redis::getHandle(token)

Syntax

getHandleStatus()

Details

The method is used to get information about all established connections. It returns a table with the following columns:

• token: The unique identifier for the connection, used to obtain the handle through getHandle(token).
• address: The "ip:port" network of the connected Redis server.
• createdTime: The time when the connection is created.

Examples

status = redis::getHandleStatus()

loadPlugin("redis");
go

dropStreamTable(`table1)

colName=["key", "value"]
colType=["string", "string"]
enableTableShareAndPersistence(table=streamTable(100:0, colName, colType), tableName=`table1, cacheSize=10000, asynWrite=false)

def myHandle(conn, msg) {
	redis::batchSet(conn, msg[0], msg[1])
}

conn = redis::connect("replace_with_redis_server_ip",6379)
subscribeTable(tableName="table1", handler=myHandle{conn})

n = 1000000
for(x in 0:n){
	insert into table1 values("key" + x, "value" + x)
}

t = table(n:0, [`id, `val], [`string, `string])
for(x in 0:n){
	insert into t values("key" + x, redis::run(conn, "GET", "key" + x))
}

ret = exec count(*) from t

assert "test", n==ret

Load the redis plugin and configure basic information for consumption.

loadPlugin("redis")

host = "127.0.0.1"
port = 6379
streamKey = "test_stream"
groupName = "test_group"
consumerName = "consumer1"

Create a Redis connection and consumer group.

conn = redis::connect(host, port)

// Delete the stream if it exists
redis::run(conn, "DEL", streamKey)

/*
msgId="0": Consume from the beginning.
mkStream=true: Create the stream if it does not exist.
*/ 
redis::createStreamGroup(conn, streamKey, groupName, "0", true)

Insert three messages into the Redis stream.

msgId1 = redis::run(conn, "XADD", streamKey, "*", "time", string(now()), "msg", string("XADD_MSG" + rand(10000000000)[0]))
msgId2 = redis::run(conn, "XADD", streamKey, "*", "time", string(now()), "msg", string("XADD_MSG" + rand(10000000000)[0]))
msgId3 = redis::run(conn, "XADD", streamKey, "*", "time", string(now()), "msg", string("XADD_MSG" + rand(10000000000)[0]))
// View message count in the stream
redis::run(conn, "XLEN", streamKey)

Read and acknowledge messages.

// Read new messages (readPending=false)
/*
count = 10: Read up to 10 messages each time.
readPending = false: Do not read pending messages.
maxWaitTime = 0: Do not wait for new messages.
writeToPending = true: Do not acknowledge messages automatically.
*/ 
redis::readStreamGroup(conn, streamKey, groupName, consumerName, 10, ">", 0, true)

// Read pending messages (readPending=true)
msgs = redis::readStreamGroup(conn, streamKey, groupName, consumerName, 10, "0")
// Acknowledge messages
redis::ackMessages(conn, streamKey, groupName, keys(msgs))
// Read pending messages again (the result should be empty)
redis::readStreamGroup(conn, streamKey, groupName, consumerName, 10, "0")

Load the redis plugin and insert data into the Redis stream.

loadPlugin("redis")

// Insert data into the stream in a separate session
REDIS_HOST = "127.0.0.1"
REDIS_PORT = 6379
STREAM_KEY = "test_stream_sub"
GROUP_NAME = "test_group"
CONSUMER_NAME = "consumer1"
ACTION_NAME = "test_action"
conn = redis::connect(REDIS_HOST, REDIS_PORT)
for (i in 1..100000) {
    redis::run(conn, "XADD", STREAM_KEY, "*", "field1", "value_" + string(i), "field2", string(i * 100))
    sleep(10)
}

Configure the subscription information.

REDIS_HOST = "127.0.0.1"
REDIS_PORT = 6379
STREAM_KEY = "test_stream_sub"
GROUP_NAME = "test_group"
CONSUMER_NAME = "consumer1"
ACTION_NAME = "test_action"
conn = redis::connect(REDIS_HOST, REDIS_PORT)
// Clear test data (optional)
// redis::run(conn, "DEL", STREAM_KEY)
// redis::run(conn, "XGROUP", "DESTROY", STREAM_KEY, GROUP_NAME)

Create the consumer group, and define the output table and parser function.

// Create the consumer group
redis::createStreamGroup(conn, STREAM_KEY, GROUP_NAME, "0", true)

// Define the output table
share table(100:0, `msgId`field1`field2`timestamp, [STRING, STRING, STRING, TIMESTAMP]) as outputTable

// Define the parser function for converting the dictionary into a table
def msgParser(msgs) {
    msgIds = msgs.keys()
    n = msgIds.size()
    if (n == 0) {
        return table(0:0, `msgId`field1`field2`timestamp, [STRING, STRING, STRING, TIMESTAMP])
    }
    result = table(n:0, `msgId`field1`field2`timestamp, [STRING, STRING, STRING, TIMESTAMP])
    for (msgId in msgIds) {
        fields = msgs[msgId]
        field1 = fields["field1"]
        field2 = fields["field2"]
        insert into result values(msgId, field1, field2, now())
    }
    return result
}

Subscribe to the stream, view subscription results, unsubscribe to the stream, and clear the environment.

// Subscribe to the stream
redis::subscribeStream(REDIS_HOST, REDIS_PORT, STREAM_KEY, GROUP_NAME, CONSUMER_NAME,
                       outputTable, msgParser, ACTION_NAME, ">", 100, true)

// View subscription results
redis::getSubscriptionStats()
select * from outputTable

// Unsubscribe to the stream
redis::unsubscribeStream(ACTION_NAME)
redis::getSubscriptionStats()

// Clear the environment
redis::release(conn)
undef(`outputTable, SHARED);