首先是程序的自定義，這里設置了redis插件需要的參數，默認值，以及校驗等。

然后注冊Redis實例需要的信息，比如key的名字或者url等，可以看到默認的data_type是list模式。

程序運行的主要入口，根據不同的data_type，傳遞不同的實現方法，然后調用listener_loop執行循環監聽

Listner_loop方法傳遞了兩個參數，一個是監聽器實現的方法，一個是處理的數據隊列。循環是每秒鐘執行一次，如果循環標識被設置，則退出。

上面的循環方法可以看到，是通過一個參數shutdown_requested來判斷是否繼續循環。該參數通過tear_down方法設置為true，然后根據不同的模式，指定不同的退出方式。
如果是list模式，則直接退出；如果是channel模式，則發送redis的unsubsribe命令退出；如果是pattern_channel,則發送punsubscribe退出。

在循環內部，判斷是否已經創建了redis實例，如果沒有創建，則調用connect方法創建；否則直接執行。

這里前一段是調用Redis的new方法，初始化一個redis實例。緊接著判斷batch_count是否大于1，如果等于1，就什么也不做，然后返回redis。
如果batch_count大于1，那么就調用load_batch_script方法，加載Lua腳本，存儲到redis中的lua腳本字典中，供后面使用。代碼如下：

上面的代碼應該是這個插件最難理解的部分了。為了弄清楚這段代碼的工作，需要了解下面幾個知識點：

• lua腳本基本概念
• Redis中的EVAL命令如何使用
• 理解上面腳本的工作

首先，要想運行上面的腳本，必須是Redis2.6+的版本，才支持EVAL，否則會報錯！EVAL命令與js中的差不多，就是可以把某一個字符串當做命令解析，其中字符串就包括lua腳本。這樣有什么好處呢？

說白了，就是能一次性進行多個操作。比如我們可以在腳本中寫入一連串的操作，這些操作會以原子模式，一次性在服務器執行完，在返回回來。

Lua腳本

關于lua腳本，其實沒有詳細研究的必要，但是一定要知道一個local和table的概念。local是創建本地的變量，這樣就不會污染redis的數據。table是lua的一種數據結構，有點類似于json，可以存儲數據。

EVAL命令

另外還要知道EVAL命令的使用方法，看下面這個命令，就好理解了！
EVAL "return KEYS[1] KEYS[2] ARGV[1] ARGV[2];" 2 name:xing age:13
就會返回：

name
age
xing
13

這段代碼沒有經過真正的操作，但是有助于理解就好！也就是說，EVAL后面跟著一段腳本，腳本后面跟著的就是參數，可以通過KEYS和ARGV數組獲得，但是下標從1開始。

再來說說EVAL命令，它的執行過程如下：

• 解析字符串腳本，根據校驗和生成lua的方法
• 把校驗和和函數放入一個lua_script字典里面，之后就可以通過EVALSHA命令直接使用校驗和執行函數。

有了這些理論基礎以后，就可以看看上面的代碼都做了什么了！
首先是獲取參數，這個參數賦值給i；然后創建了一個對象res；緊接著調用llen命令，獲得指定list的長度；如果list的長度大于i，則什么也不做；如果小于i，那么i就等于lenth；然后執行命令lpop，取出list中的元素，一共取i次，放入res中，最后返回。

說得通俗點，就是比較一下list元素個數與設置batch_count的值。如果batch_count為5，列表list中有5條以上的數據，那么直接取5條，一次性返回；否則取length條返回。

可以看到這段腳本的作用，就是讓logstash一次請求，最多獲得batch_count條事件，減小了服務器處理請求的壓力。

講完這段代碼，可以看看不同的工作模式的實現代碼了：

首先是list的代碼，其實就是執行BLPOP命令，獲取數據。如果在list模式中，還會去判斷batch_count的值，如果是1直接退出；如果大于1，則使用evalsha命令調用之前保存的腳本方法。

至于channel和pattern_channel,就沒啥解釋的了，就是分別調用subscribe和psubsribe命令而已。

其實最難理解的，就是中間那段lua腳本~明白它的用處，redis插件也就不難理解了。

代碼

# encoding: utf-8
require "logstash/inputs/base"
require "logstash/inputs/threadable"
require "logstash/namespace"# This input will read events from a Redis instance; it supports both Redis channels and lists.
# The list command (BLPOP) used by Logstash is supported in Redis v1.3.1+, and
# the channel commands used by Logstash are found in Redis v1.3.8+.
# While you may be able to make these Redis versions work, the best performance
# and stability will be found in more recent stable versions.  Versions 2.6.0+
# are recommended.
#
# For more information about Redis, see <http://redis.io/>
#
# `batch_count` note: If you use the `batch_count` setting, you *must* use a Redis version 2.6.0 or
# newer. Anything older does not support the operations used by batching.
#
class LogStash::Inputs::Redis < LogStash::Inputs::Threadableconfig_name "redis"default :codec, "json"# The `name` configuration is used for logging in case there are multiple instances.# This feature has no real function and will be removed in future versions.config :name, :validate => :string, :default => "default", :deprecated => true# The hostname of your Redis server.config :host, :validate => :string, :default => "127.0.0.1"# The port to connect on.config :port, :validate => :number, :default => 6379# The Redis database number.config :db, :validate => :number, :default => 0# Initial connection timeout in seconds.config :timeout, :validate => :number, :default => 5# Password to authenticate with. There is no authentication by default.config :password, :validate => :password# The name of the Redis queue (we'll use BLPOP against this).# TODO: remove soon.config :queue, :validate => :string, :deprecated => true# The name of a Redis list or channel.# TODO: change required to trueconfig :key, :validate => :string, :required => false# Specify either list or channel.  If `redis\_type` is `list`, then we will BLPOP the# key.  If `redis\_type` is `channel`, then we will SUBSCRIBE to the key.# If `redis\_type` is `pattern_channel`, then we will PSUBSCRIBE to the key.# TODO: change required to trueconfig :data_type, :validate => [ "list", "channel", "pattern_channel" ], :required => false# The number of events to return from Redis using EVAL.config :batch_count, :validate => :number, :default => 1publicdef registerrequire 'redis'@redis = nil@redis_url = "redis://#{@password}@#{@host}:#{@port}/#{@db}"# TODO remove after setting key and data_type to trueif @queueif @key or @data_typeraise RuntimeError.new("Cannot specify queue parameter and key or data_type")end@key = @queue@data_type = 'list'endif not @key or not @data_typeraise RuntimeError.new("Must define queue, or key and data_type parameters")end# end TODO@logger.info("Registering Redis", :identity => identity)end # def register# A string used to identify a Redis instance in log messages# TODO(sissel): Use instance variables for this once the @name config# option is removed.privatedef identity@name || "#{@redis_url} #{@data_type}:#{@key}"endprivatedef connectredis = Redis.new(:host => @host,:port => @port,:timeout => @timeout,:db => @db,:password => @password.nil? ? nil : @password.value)load_batch_script(redis) if @data_type == 'list' && (@batch_count > 1)return redisend # def connectprivatedef load_batch_script(redis)#A Redis Lua EVAL script to fetch a count of keys#in case count is bigger than current items in queue whole queue will be returned without extra nil valuesredis_script = <<EOFlocal i = tonumber(ARGV[1])local res = {}local length = redis.call('llen',KEYS[1])if length < i then i = length endwhile (i > 0) dolocal item = redis.call("lpop", KEYS[1])if (not item) thenbreakendtable.insert(res, item)i = i-1endreturn res
EOF@redis_script_sha = redis.script(:load, redis_script)endprivatedef queue_event(msg, output_queue)begin@codec.decode(msg) do |event|decorate(event)                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                               output_queue << eventendrescue LogStash::ShutdownSignal => e# propagate upraise(e)rescue => e # parse or event creation error@logger.error("Failed to create event", :message => msg, :exception => e, :backtrace => e.backtrace);endendprivatedef list_listener(redis, output_queue)item = redis.blpop(@key, 0, :timeout => 1)return unless item # from timeout or other conditions# blpop returns the 'key' read from as well as the item result# we only care about the result (2nd item in the list).queue_event(item[1], output_queue)# If @batch_count is 1, there's no need to continue.return if @batch_count == 1beginredis.evalsha(@redis_script_sha, [@key], [@batch_count-1]).each do |item|queue_event(item, output_queue)end# Below is a commented-out implementation of 'batch fetch'# using pipelined LPOP calls. This in practice has been observed to# perform exactly the same in terms of event throughput as# the evalsha method. Given that the EVALSHA implementation uses# one call to Redis instead of N (where N == @batch_count) calls,# I decided to go with the 'evalsha' method of fetching N items# from Redis in bulk.#redis.pipelined do#error, item = redis.lpop(@key)#(@batch_count-1).times { redis.lpop(@key) }#end.each do |item|#queue_event(item, output_queue) if item#end# --- End commented out implementation of 'batch fetch'rescue Redis::CommandError => eif e.to_s =~ /NOSCRIPT/ then@logger.warn("Redis may have been restarted, reloading Redis batch EVAL script", :exception => e);load_batch_script(redis)retryelseraise eendendendprivatedef channel_listener(redis, output_queue)redis.subscribe @key do |on|on.subscribe do |channel, count|@logger.info("Subscribed", :channel => channel, :count => count)endon.message do |channel, message|queue_event message, output_queueendon.unsubscribe do |channel, count|@logger.info("Unsubscribed", :channel => channel, :count => count)endendendprivatedef pattern_channel_listener(redis, output_queue)redis.psubscribe @key do |on|on.psubscribe do |channel, count|@logger.info("Subscribed", :channel => channel, :count => count)endon.pmessage do |ch, event, message|queue_event message, output_queueendon.punsubscribe do |channel, count|@logger.info("Unsubscribed", :channel => channel, :count => count)endendend# Since both listeners have the same basic loop, we've abstracted the outer# loop.privatedef listener_loop(listener, output_queue)while !@shutdown_requestedbegin@redis ||= connectself.send listener, @redis, output_queuerescue Redis::BaseError => e@logger.warn("Redis connection problem", :exception => e)# Reset the redis variable to trigger reconnect@redis = nilsleep 1endendend # listener_looppublicdef run(output_queue)if @data_type == 'list'listener_loop :list_listener, output_queueelsif @data_type == 'channel'listener_loop :channel_listener, output_queueelselistener_loop :pattern_channel_listener, output_queueendrescue LogStash::ShutdownSignal# ignore and quitend # def runpublicdef teardown@shutdown_requested = trueif @redisif @data_type == 'list'@redis.quit rescue nilelsif @data_type == 'channel'@redis.unsubscribe rescue nil@redis.connection.disconnectelsif @data_type == 'pattern_channel'@redis.punsubscribe rescue nil@redis.connection.disconnectend@redis = nilendend
end # class LogStash::Inputs::Redis

?