lua-resty-rocketmq - Lua rocketmq client driver for the ngx_lua based on the cosocket API

• Name
• Status
• Description
• Quick start
• Synopsis
• Modules
  • resty.rocketmq.producer
    • Methods
      • new
      • addRPCHook
      • setUseTLS
      • setTimeout
      • registerSendMessageHook
      • registerEndTransactionHook
      • send
      • sendMessageInTransaction
      • batchSend
      • start
      • stop
  • resty.rocketmq.consumer
    • Methods
      • new
      • addRPCHook
      • setUseTLS
      • setTimeout
      • registerMessageListener
      • registerConsumeMessageHook
      • subscribe
      • start
      • stop
      • setAllocateMessageQueueStrategy
      • getAllocateMessageQueueStrategy
      • setEnableMsgTrace
      • isEnableMsgTrace
      • setCustomizedTraceTopic
      • getCustomizedTraceTopic
      • setConsumeFromWhere
      • getConsumeFromWhere
      • setConsumeTimestamp
      • getConsumeTimestamp
      • setPullThresholdForQueue
      • getPullThresholdForQueue
      • setPullThresholdSizeForQueue
      • getPullThresholdSizeForQueue
      • setPullTimeDelayMillsWhenException
      • getPullTimeDelayMillsWhenException
      • setPullBatchSize
      • getPullBatchSize
      • setPullInterval
      • getPullInterval
      • setConsumeMessageBatchMaxSize
      • getConsumeMessageBatchMaxSize
      • setMaxReconsumeTimes
      • getMaxReconsumeTimes
  • resty.rocketmq.admin
    • Methods
      • new
      • addRPCHook
      • setUseTLS
      • setTimeout
      • createTopic
      • createTopicForBroker
      • searchOffset
      • maxOffset
      • minOffset
      • earliestMsgStoreTime
      • viewMessage
      • queryMessage
      • queryTraceByMsgId
• Installation
• See Also

Production ready.

This Lua library is a RocketMq client driver for the ngx_lua nginx module:

This Lua library takes advantage of ngx_lua's cosocket API, which ensures 100% nonblocking behavior.

for ubuntu:

wget -O - https://openresty.org/package/pubkey.gpg | sudo apt-key add -
echo "deb http://openresty.org/package/ubuntu $(lsb_release -sc) main" \
    | sudo tee /etc/apt/sources.list.d/openresty.list
sudo apt-get update
sudo apt-get -y install openresty

see https://openresty.org/cn/linux-packages.html for more distributions

wget https://archive.apache.org/dist/rocketmq/4.9.3/rocketmq-all-4.9.3-bin-release.zip
unzip rocketmq-all-4.9.3-bin-release.zip
cd rocketmq-4.9.3
nohup bash bin/mqnamesrv &
nohup bash bin/mqbroker -n localhost:9876 -c conf/broker.conf &

cd examples
chmod +x producer.lua
./producer.lua

    lua_package_path "/path/to/lua-resty-rocketmq/lib/?.lua;;";

    server {
        location /test {
            content_by_lua_block {
                local cjson = require "cjson"
                local producer = require "resty.rocketmq.producer"
                local consumer = require "resty.rocketmq.consumer"

                local nameservers = { "127.0.0.1:9876" }

                local message = "halo world"

                local p = producer.new(nameservers, "produce_group")
                
                -- set acl
                local aclHook = require("resty.rocketmq.acl_rpchook").new("RocketMQ","123456781")
                p:addRPCHook(aclHook)
                
                -- use tls mode
                p:setUseTLS(true)
                
                local res, err = p:send("TopicTest", message)
                if not res then
                    ngx.say("send err:", err)
                    return
                end
                ngx.say("send success")
                
                -- consume
                local c = consumer.new(nameservers, "group1")
                c:subscribe("TopicTest", "*")
                c:registerMessageListener({
                    consumeMessage = function(self, msgs, context)
                        ngx.say("consume success:", cjson.encode(msgs))
                        return consumer.CONSUME_SUCCESS
                    end
                })
                
                c:start()
                ngx.sleep(5)
                c:stop()
            }
        }
    }

Back to TOC

To load this module, just do this

    local producer = require "resty.rocketmq.producer"

syntax: p = producer.new(nameservers, produce_group, enableMsgTrace, customizedTraceTopic)

nameservers is list of nameserver addresses

syntax: p:addRPCHook(hook)

hook is a table that contains two functions as follows:

• doBeforeRequest(self, addr, header, body)
• doAfterResponse(self, addr, header, body, respHeader, respBody)

there is an acl hook provided, usage is:

    local accessKey, secretKey = "RocketMQ", "12345678"
    local aclHook = require("resty.rocketmq.acl_rpchook").new(accessKey, secretKey)
    p:addRPCHook(aclHook)

syntax: p:setUseTLS(useTLS)

useTLS is a boolean

syntax: p:setTimeout(timeout)

timeout is in milliseconds, default 3000

syntax: p:registerSendMessageHook(hook)

hook is a table that contains two functions as follows:

• sendMessageBefore(self, context)
• sendMessageAfter(self, context)

context is a table that contains:

• producer
• producerGroup
• communicationMode
• bornHost
• brokerAddr
• message
• mq
• msgType
• sendResult
• exception

syntax: p:registerEndTransactionHook(hook)

hook is a table that contains a function as follows:

• endTransaction(self, context)

context is a table that contains:

• producerGroup
• brokerAddr
• message
• msgId
• transactionId
• transactionState
• fromTransactionCheck

syntax: res, err = p:send(topic, message, tags, keys, properties)

properties is a table that contains:

• waitStoreMsgOk
• delayTimeLevel

In case of success, returns the a table of results. In case of errors, returns nil with a string describing the error.

syntax: res, err = p:setTransactionListener(transactionListener)

transactionListener is a table that contains two functions as follows:

• executeLocalTransaction(self, msg, arg)
• checkLocalTransaction(self, msg)

syntax: res, err = p:sendMessageInTransaction(topic, arg, message, tags, keys, properties)

syntax: res, err = p:batchSend(msgs)

msgs is a list of msgs, each msg is a table that contains:

• topic
• body
• tags
• keys
• properties

syntax: p:start()

note that if you don't call p:start() before sending messages, messages will be sent successfully, but the trace is not send.

syntax: p:stop()

Back to TOC

To load this module, just do this

    local consumer = require "resty.rocketmq.consumer"

syntax: c = consumer.new(nameservers, consumerGroup)

nameservers is list of nameserver addresses

syntax: c:addRPCHook(hook)

hook is a table that contains two functions as follows:

• doBeforeRequest(self, addr, header, body)
• doAfterResponse(self, addr, header, body, respHeader, respBody)

there is an acl hook provided, usage is:

    local accessKey, secretKey = "RocketMQ", "12345678"
    local aclHook = require("resty.rocketmq.acl_rpchook").new(accessKey, secretKey)
    c:addRPCHook(aclHook)

syntax: c:setUseTLS(useTLS)

useTLS is a boolean

syntax: c:setTimeout(timeout)

timeout is in milliseconds, default 3000

syntax: c:registerMessageListener(messageListener)

messageListener is a table that contains a function as follows:

• consumeMessage(self, msgs, context)

syntax: c:registerConsumeMessageHook(hook)

hook is a table that contains two functions as follows:

• consumeMessageBefore(self, context)
• consumeMessageAfter(self, context)

context is a table that contains:

• consumerGroup
• mq
• msgList
• success
• status
• consumeContextType

syntax: c:subscribe(topic, subExpression)

syntax: c:start()

syntax: c:stop()

syntax: c:setAllocateMessageQueueStrategy(strategy)

strategy is a functions as follows:

• function(consumerGroup, currentCID, mqAll, cidAll)

default value is consumer.AllocateMessageQueueAveragely

syntax: local strategy = c:getAllocateMessageQueueStrategy()

syntax: c:setEnableMsgTrace(enableMsgTrace)

syntax: local enableMsgTrace = c:isEnableMsgTrace()

syntax: c:setCustomizedTraceTopic(customizedTraceTopic)

syntax: local customizedTraceTopic = c:getCustomizedTraceTopic()

syntax: c:setConsumeFromWhere(consumeFromWhere)

syntax: local consumeFromWhere = c:getConsumeFromWhere()

syntax: c:setConsumeTimestamp(consumeTimestamp)

syntax: local consumeTimestamp = c:getConsumeTimestamp()

syntax: c:setPullThresholdForQueue(pullThresholdForQueue)

syntax: local pullThresholdForQueue = c:getPullThresholdForQueue()

syntax: c:setPullThresholdSizeForQueue(pullThresholdSizeForQueue)

syntax: local pullThresholdSizeForQueue = c:getPullThresholdSizeForQueue()

syntax: c:setPullTimeDelayMillsWhenException(pullTimeDelayMillsWhenException)

syntax: local pullTimeDelayMillsWhenException = c:getPullTimeDelayMillsWhenException()

syntax: c:setPullBatchSize(pullBatchSize)

syntax: local pullBatchSize = c:getPullBatchSize()

syntax: c:setPullInterval(pullInterval)

syntax: local pullInterval = c:getPullInterval()

syntax: c:setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize)

syntax: local consumeMessageBatchMaxSize = c:getConsumeMessageBatchMaxSize()

syntax: c:setMaxReconsumeTimes(maxReconsumeTimes)

syntax: local maxReconsumeTimes = c:getMaxReconsumeTimes()

syntax: c:setClientRebalance(clientRebalance)

syntax: local clientRebalance = c:isClientRebalance()

syntax: c:setPopThresholdForQueue(popThresholdForQueue)

syntax: local popThresholdForQueue = c:getPopThresholdForQueue()

syntax: c:setPopInvisibleTime(popInvisibleTime)

syntax: local popInvisibleTime = c:getPopInvisibleTime()

syntax: c:setPopBatchNums(popBatchNums)

syntax: local popBatchNums = c:getPopBatchNums()

Back to TOC

To load this module, just do this

    local admin = require "resty.rocketmq.admin"

syntax: adm = admin.new(nameservers)

nameservers is list of nameserver addresses

syntax: adm:addRPCHook(hook)

hook is a table that contains two functions as follows:

• doBeforeRequest(self, addr, header, body)

• doAfterResponse(self, addr, header, body, respHeader, respBody)

there is an acl hook provided, usage is:

    local accessKey, secretKey = "RocketMQ", "12345678"
    local aclHook = require("resty.rocketmq.acl_rpchook").new(accessKey, secretKey)
    adm:addRPCHook(aclHook)

syntax: adm:setUseTLS(useTLS)

useTLS is a boolean

syntax: adm:setTimeout(timeout)

timeout is in milliseconds, default 3000

syntax: res, err = adm:createTopic(newTopic, queueNum, topicSysFlag)

• newTopic: the new topic name
• queueNum: read and write queue numbers
• topicSysFlag: system flag of the topic

syntax: res, err = adm:createTopicForBroker(addr, topicConfig)

• addr: broker address
• topicConfig: a table containing:
  • topic
  • readQueueNums
  • writeQueueNums
  • perm
  • topicFilterType
  • topicSysFlag
  • order

syntax: res, err = adm:searchOffset(mq, timestamp)

• mq: a table containing:
  • brokerName
  • topic
  • queueId
• timestamp: search time

syntax: res, err = adm:maxOffset(mq)

• mq: a table containing:
  • brokerName
  • topic
  • queueId

syntax: res, err = adm:minOffset(mq)

• mq: a table containing:
  • brokerName
  • topic
  • queueId

syntax: res, err = adm:earliestMsgStoreTime(mq)

• mq: a table containing:
  • brokerName
  • topic
  • queueId

syntax: res, err = adm:viewMessage(offsetMsgId)

syntax: res, err = adm:queryMessage(topic, key, maxNum, beginTime, endTime, isUniqKey)

syntax: res, err = adm:queryTraceByMsgId(traceTopic, msgId)

Back to TOC

Proxy provides http API of produce/consume messages.

Install and start RocketMQ

wget https://archive.apache.org/dist/rocketmq/4.9.3/rocketmq-all-4.9.3-bin-release.zip
unzip rocketmq-all-4.9.3-bin-release.zip
cd rocketmq-4.9.3
nohup bash bin/mqnamesrv &
nohup bash bin/mqbroker -n localhost:9876 -c conf/broker.conf &

Start proxy

cd examples/
openresty -c server/proxy.conf -p .

Send message using proxy

curl localhost:8080/topics/topic1/messages -d '{"properties": {"a": "3", "KEYS": "key", "TAGS": "tag"}, "body": "hello proxy"}'

Back to TOC

You need to configure the lua_package_path directive to add the path of your lua-resty-rocketmq source tree to ngx_lua's LUA_PATH search path, as in

    # nginx.conf
    http {
        lua_package_path "/path/to/lua-resty-rocketmq/lib/?.lua;;";
        ...
    }

Ensure that the system account running your Nginx ''worker'' proceses have enough permission to read the .lua file.

Back to TOC

• ngx_lua module: https://github.com/openresty/lua-nginx-module
• apache rocketmq: https://github.com/apache/rocketmq
• lua-resty-kafka: https://github.com/doujiang24/lua-resty-kafka
• luatz: https://github.com/daurnimator/luatz

Back to TOC