我们之前写过一篇logstash收集数据输出到redis的笔记：https://sulao.cn/post/736，可能我们有一点顾虑，redis单个key的最大value大小为512M，

在生产环境中收集集群机器的日志数据可能512M一会就会写满，所以写满以后导致了redis的数据积压，导致数据读写瓶颈，其实logstash写入和读取redis的时候使用list方式分别使用rpush和lblpop推入和删除已读数据，logstash插入数据到list的最后，logstash不断取走列表的第一个组数据并删除，redis读取数据非常快，所以一般情况下很难导致redis的数据积压

我们可以考虑将logstash中redis配置的data_type改为channel/pattern_channel，这样虽然能够避免可能导致redis数据积压的问题，但是当ES断线的时候导致读取不到断线时的消息，因为redis不支持重复消费订阅频道的信息

我们来看看配置例子

#cat /etc/logstash/conf.d/logstash.conf
input {
  beats {
    port => 5044
    codec => plain {
          charset => "UTF-8"
    }
  }
}

filter {
  grok {
    match => ["message", '%{IP:Client} - - \[%{HTTPDATE:timestamp}\] \"(%{WORD:method} %{NOTSPACE:request} HTTP/%{NUMBER:httpversion})\" %{NUMBER:response} (?:%{NUMBER:bytes}|-) %{QS:referrer} %{QS:agent}']
  }
  if "beats_input_codec_plain_applied" in [tags] {
    mutate {
      remove_tag => ["beats_input_codec_plain_applied"]
    }
  }
}

#output {
#  elasticsearch {
#    hosts => "172.26.61.61:9200"
#    manage_template => false
#    index => "%{[@metadata][beat]}-%{+YYYY.MM.dd}"
#  }
#}

output {
  redis {
    data_type => "channel" #工作方式
    host => "172.26.61.58"
    port => "6379"
    db => "0"
    #password => "123456"
    key => "nginx_log"
  }
}

#cat /etc/logstash/conf.d/redis.conf
input {
  redis {
    data_type => "channel" #工作方式
    host => "172.26.61.58"
    port => "6379"
    db => "0"
    #password => "123456"
    key => "nginx_log"
    threads => 1
  }
}

output {
  elasticsearch {
    hosts => "172.26.61.61:9200"
    manage_template => false
    index => "%{[@metadata][beat]}-%{+YYYY.MM.dd}"
  }
}

这种方式在需要扩展 logstash 成多节点集群的时候，会出现一个问题：通过频道发布的一条信息，会被所有订阅了该频道的 logstash 进程同时接收到，然后输出重复内容！