RabbitMQ

• DolphinDB Server: 2.00.10及更高版本
• OS: 64位 Linux

1. 在DolphinDB 客户端中使用 listRemotePlugins 命令查看插件仓库中的插件信息。

   login("admin", "123456")
   listRemotePlugins()

2. 使用 installPlugin 命令完成插件安装。

   installPlugin("rabbitmq")

3. 使用 loadPlugin 命令加载插件。

   loadPlugin("rabbitmq")

rabbitmq::createCSVFormatter([format], [delimiter=','], [rowDelimiter=';'])

创建一个CSV格式的Formatter函数。

参数：

• 'format': 是一个STRING类型的向量。
• 'delimiter': 是列之间的分隔符，默认是','。
• 'rowDelimiter': 是行之间的分隔符，默认是';'。

例子：

MyFormat = take("", 3)
MyFormat[1] = "0.000"
formatter = rabbitmq::createCSVFormatter(MyFormat,',',';')

rabbitmq::createCSVParser(schema, [delimiter=','], [rowDelimiter=';'])

创建一个CSV格式的Parser函数。

参数：

• 'schema': 是一个包含各列数据类型的向量。
• 'delimiter': 是列之间的分隔符，默认是','。
• 'rowDelimiter': 是行之间的分隔符，默认是';'。

例子：

parser = rabbitmq::createCSVParser([INT,STRING,INT], ',',';')

rabbitmq::createProtoBuformatter()

创建一个protobuf格式的Formatter函数。

参数：无。

例子：

formatter = rabbitmq::createProtoBuformatter()

rabbitmq::createProtoBufParser()

创建一个JSON格式的Parser函数。

参数：无

例子：

parser = rabbitmq::createProtoBufParser()

rabbitmq::createJSONFormatter()

创建一个JSON格式的Formatter函数。

参数：无。

例子：

formatter = rabbitmq::createJSONFormatter()

rabbitmq::createJSONParser(schema, colNames)

创建一个JSON格式的Parser函数。

参数：

• 'schema' 是一个向量，表示各列的数据类型。
• 'colNames' 是一个向量，表示列名。

例子：

parser = rabbitmq::createJSONParser([INT,TIMESTAMP,INT], ["id","ts","volume"])

rabbitmq::close(handle)

关闭一个connection或者一个channel，注意，如果关闭connection，其上的channel也会自动关闭

参数：

• 'handle' 是一个handle, 是由connection接口或者channel接口返回的

例子：

conn = rabbitmq::connection('localhost', 5672, 'guest', 'guest', "/")
rabbitmq::close(conn)

rabbitmq::closeAll()

关闭所有已经建立的connection和channel。

例子：

rabbitmq::closeAll()

语法

rabbitmq::getConnections()

详情

获取创所有已创建的 RabbitMQ 连接信息。

返回一个元组，每个元素都是一个字典，其键值如下：

• connection：连接句柄。
• createTime：创建连接的时间，类型为 TIMESTAMP。

语法

rabbitmq::getChannels([connection])

详情

如果填写了 connection 参数，则查询通过该 connection 所创建的通道信息；否则则查询所有已创建的通道信息。

返回一个元组，每个元素都是一个字典，其键值如下：

• connection：通道句柄所对应的连接句柄。
• channel：通道句柄。
• createTime：表示创建通道的时间，类型为 TIMESTAMP。

参数

connection 连接句柄。

// 加载插件
loadPlugin("/home/yxu/Desktop/DolphinDBPlugin/rabbitmq/build/PluginRabbitMQ.txt")
// 建立连接
conn = rabbitmq::connection('localhost', 5672, 'guest', 'guest', '');
// 创建通道
ch = rabbitmq::channel(conn)
// 声明Exchange
rabbitmq::declareExchange(ch, 'test-exchange', 'fanout', ['durable'])
rabbitmq::declareExchange(ch, 'test-exchange1', 'fanout', ['durable'])
rabbitmq::declareExchange(ch, 'test-exchange2', 'fanout', ['durable'])
rabbitmq::declareExchange(ch, 'test-exchange3', 'fanout', ['durable'])
rabbitmq::declareExchange(ch, 'test-exchange4', 'fanout', ['durable'])
// 绑定Exchange
rabbitmq::bindExchange(ch, 'test-exchange1', 'test-exchange2', 'rule1')
// 解绑Exchange
rabbitmq::unbindExchange(ch, 'test-exchange1', 'test-exchange2', 'rule1')
// 移除Exchange
rabbitmq::removeExchange(ch, 'test-exchange')
rabbitmq::removeExchange(ch, 'test-exchange1')
rabbitmq::removeExchange(ch, 'test-exchange2')
rabbitmq::removeExchange(ch, 'test-exchange3')
rabbitmq::removeExchange(ch, 'test-exchange4')
// 声明Queue
rabbitmq::declareQueue(ch, 'test-queue1', ['durable'])
rabbitmq::declareQueue(ch, 'test-queue2', ['durable'])
rabbitmq::declareQueue(ch, 'test-queue3', ['durable'])
rabbitmq::declareQueue(ch, 'test-queue4', ['durable'])
// 绑定Queue
rabbitmq::bindQueue(ch, 'test-exchange1', 'test-queue1', 'rule1')
rabbitmq::bindQueue(ch, 'test-exchange2', 'test-queue2', 'rule1')
rabbitmq::bindQueue(ch, 'test-exchange3', 'test-queue3', 'rule1')
rabbitmq::bindQueue(ch, 'test-exchange4', 'test-queue4', 'rule1')
// 解绑Queue
rabbitmq::unbindQueue(ch, 'test-exchange1', 'test-queue', 'rule1')
// 移除Queue
rabbitmq::removeQueue(ch, 'test-queue1')
rabbitmq::removeQueue(ch, 'test-queue2')
rabbitmq::removeQueue(ch, 'test-queue3')
rabbitmq::removeQueue(ch, 'test-queue4')
// 发数据
rabbitmq::publish(ch, 'test-exchange1', 'rule1', 'Hello World1', 'bytestream')
rabbitmq::publish(ch, 'test-exchange2', 'rule1', 'Hello World1', 'bytestream')
// 取数据
def handle(msg){
}
rabbitmq::consume(ch, 'test-queue1', handle, 'bytestream', 'consumer1')

def handle(msg){
}
rabbitmq::consume(ch, 'test-queue2', handle, 'bytestream', 'consumer2')

// 取消取数据
rabbitmq::cancelConsume(ch, 'consumer1')
rabbitmq::cancelConsume(ch, 'consumer2')

// 发送json格式数据
formatter = rabbitmq::createJSONFormatter()
data = table(1..10 as id, take(now(), 10) as ts, rand(10, 10) as volume)
rabbitmq::publish(ch, 'test-exchange3', 'rule1', data, formatter)

// 发送csv格式数据
MyFormat = take("", 3)
MyFormat[1] = "0.000"
print(MyFormat)
data = table(1..10 as id, take(now(), 10) as ts, rand(10, 10) as volume)
formatter = rabbitmq::createCSVFormatter(MyFormat,',',';')
rabbitmq::publish(ch, 'test-exchange4', 'rule1', data, formatter)

// 接收json格式数据
parser = rabbitmq::createJSONParser([INT,TIMESTAMP,INT], ["id","ts","volume"])
def handle(msg){
}
rabbitmq::consume(ch, 'test-queue3', handle, parser, 'consumer3')

// 接收csv格式数据
parser1 = rabbitmq::createCSVParser([INT,STRING,INT], ',',';')
def handle(msg){
}
rabbitmq::consume(ch, 'test-queue4', handle, parser1, 'consumer4')

rabbitmq::cancelConsume(ch, 'consumer3')
rabbitmq::cancelConsume(ch, 'consumer4')

// 获取后台订阅信息
tb = rabbitmq::getSubJobStat()
print(tb)
// 取消某个后台订阅
rabbitmq::cancelSubJob(订阅id)

// 订阅完成GUI2可关闭
def handle(msg){
}
rabbitmq::createSubJob('test-queue2', handle, 'bytestream', 'consumer1',['localhost'], [5672], 'guest', 'guest')

// 后台订阅json格式数据
parser = rabbitmq::createJSONParser([INT,TIMESTAMP,INT], ["id","ts","volume"])
def handle(msg){
}
rabbitmq::createSubJob('test-queue3', handle, parser, 'consumer1',['localhost'], [5672], 'guest', 'guest')

// 后台订阅csv格式数据
parser1 = rabbitmq::createCSVParser([INT,STRING,INT], ',',';')
def handle(msg){
}
rabbitmq::createSubJob('test-queue4', handle, parser1, 'consumer1',['localhost'], [5672], 'guest', 'guest')