问RABBITMQ向LOGSTASH发送消息
EN

Stack Overflow用户

提问于 2021-02-11 04:02:46

回答 1查看 3.2K关注 0票数 1

你能帮我在日志中输入RabbitMQ吗？我的应用程序将代码的版本发送到rabbitmq，然后将其存储在弹性堆栈中。在rabbitmq中为app创建了队列名: app_version_queue；类型:经典；持久:真

然后用该配置配置logstash：

input {
  rabbitmq {
    id => "rabbitmyq_id"
    # connect to rabbit
    host => "localhost"
    port => 5672
    vhost => "/"
    # INPUT - PRODUCERS
    key => "app_version_queue"
    # OUTPUT - CONSUMER
    # queue for logstash
    queue => "logstash"
    auto_delete => false
    # Exchange for logstash
    exchange => logstash
    exchange_type => direct
    durable => "true"
    # No ack will boost your perf
    ack => false
  }
}

output {
  elasticsearch {
    hosts => [ "elasticsearch:9200" ]
    index => "app_version-%{+YYYY.MM.dd}"
  }
}

这是可行的，但是现在，在RabbitMQ控制台中，我在表排队的消息中看到就绪: 914,444，Unacked: 0总计: 914,444

我的磁盘空间在三天内就满了。重新启动rabbitmq服务器后，所有空间都是空闲的。

更新:所有的原因，为什么我要这么做，我想从我想做的链app=>rabbit=>nifi=>elastic : app=>rabbit=>logstash=>elastic中删除

队列: app_version我的应用程序向nifi=>ELASTIC 队列发送消息
队列: logstash，我用logstash Queue2 -对数存储创建的

我试图阻止NIFI的发送，但是消息没有离开。

rabbitmq

logstash

logstash-configuration

回答 1

Stack Overflow用户

回答已采纳

发布于 2021-02-11 08:13:44

听起来就像你已经创建了两次基础设施：

在RabbitMQ中手动一次
在LogStash的配置选项中设置一次

你需要的是--只有三样东西--

要向其发布消息的应用程序的交换。
用于使用来自LogStash的消息的队列。
该交换和该队列之间的绑定；队列将获得每个发布到交换的带有匹配路由密钥的消息的副本。

你拥有的是所有的

应用程序将消息发布到的一个名为logs (手动创建)的交换。
一个名为app_version_queue的队列(手动创建)，没有任何消耗。
一个绑定(手动创建)将消息的副本从logs传递到app_version_queue，然后app_version_queue永远在那里。
一个名为logstash的交换(由LogStash创建)，它没有向其发布消息。
一个名为logstash (由LogStash创建)的队列，LogStash从该队列中消费消息。
一个绑定(由LogStash创建)从logstash交换到logstash队列，它什么也不做，因为没有消息发布到该交换。
从logs交换到logstash队列的绑定(手动创建)，它实际上是从应用程序传递消息。

因此，对于这三种情况(交换、队列和绑定)，您需要：

决定一个名字
决定你是要创建它，还是让LogStash创建它
配置所有内容以使用相同的名称

例如，您可以保留名称logs和app_version_queue，并手动创建所有内容。

那么您的LogStash应用程序应该如下所示：

input {
  rabbitmq {
    id => "rabbitmyq_id"
    # connect to rabbit
    host => "localhost"
    port => 5672
    vhost => "/"
    # Consume from existing queue
    queue => "app_version_queue"
    # No ack will boost your perf
    ack => false
  }
}

另一方面，您可以只创建logs交换，让LogStash创建队列和绑定，如下所示：

input {
  rabbitmq {
    id => "rabbitmyq_id"
    # connect to rabbit
    host => "localhost"
    port => 5672
    vhost => "/"
    # Create a new queue
    queue => "logstash_processing_queue"
    durable => "true"
    # Take a copy of all messages with the "app_version_queue" routing key from the existing exchange
    exchange => "logs"
    key => "app_version_queue"
    # No ack will boost your perf
    ack => false
  }
}

或者，您可以让LogStash创建所有这些，并确保应用程序发布到正确的exchange：

input {
  rabbitmq {
    id => "rabbitmyq_id"
    # connect to rabbit
    host => "localhost"
    port => 5672
    vhost => "/"
    # Create a new queue
    queue => "logstash_processing_queue"
    durable => "true"
    # Create a new exchange; point your application to publish here!
    exchange => "log_exchange"
    exchange_type => "direct"
    # Take a copy of all messages with the "app_version_queue" routing key from the new exchange
    key => "app_version_queue"
    # No ack will boost your perf
    ack => false
  }
}