ActiveMQ作为Logstash的消息代理

扩展Logstash时，通常会添加一个消息代理，该消息代理用于在一个或多个Logstash节点处理传入消息之前临时缓冲传入的消息。数据通过像Beaver这样的发运人推送到代理， Beaver读取日志文件并将每个事件发送到代理。或者，应用程序可以使用Log4j附加程序之类的东西直接发送日志事件。

一个常见的选择是将Redis用作将数据存储在内存中的代理，但也可以使用其他选项（例如Apache Kafka）。有时，组织不那么热衷于引入很多新技术，并想重用现有商店。 ActiveMQ是一个广泛使用的消息传递和集成平台，它支持不同的协议，看起来非常适合用作消息代理。让我们看看将其集成的选项。

设置ActiveMQ

ActiveMQ可以使用随附的脚本轻松设置。在Linux上，只需执行./activemq console 。使用http://127.0.0.1:8161/admin/上的管理控制台，您可以创建新的队列，甚至可以将消息排队以进行测试。

管理仪表板

使用AMQP消费消息

尝试将ActiveMQ连接到Logstash的一种明显方法是使用AMQP（高级消息队列协议）。这是不同消息平台所支持的标准协议。

AMQP曾经有一个Logstash输入，但不幸的是，由于RabbitMQ是受支持的主要系统，因此它已重命名为Rabbitmq-input 。

让我们看看如果尝试将输入与ActiveMQ一起使用会发生什么。

input {rabbitmq {host => "localhost"queue => "TestQueue"port => 5672}
}output {stdout {codec => "rubydebug"}
}

我们告诉Logstash在名为TestQueue的队列的标准端口上的localhost上侦听。结果应转储到标准输出中。不幸的是，Logstash仅发出错误，因为它无法连接。

Logstash startup completed
RabbitMQ connection error: . Will reconnect in 10 seconds... {:level=>:error}

在ActiveMQ日志中，我们可以看到我们的参数正确，但是不幸的是，两个系统似乎都说出了不同的AMQP方言。

WARN | Connection attempt from non AMQP v1.0 client. AMQP,0,0,9,1
org.apache.activemq.transport.amqp.AmqpProtocolException: Connection from client using unsupported AMQP attempted
...

不幸的是，这个选项。

使用STOMP消费消息

适当命名的“ 面向简单文本的消息传递协议”是ActiveMQ支持的另一个选项。幸运的是有一个专用的输入。默认情况下，它不包含在Logstash中，但可以轻松安装。

bin/plugin install logstash-input-stomp

之后，我们可以在Logstash配置中使用它。

input {stomp {host => "localhost"destination => "TestQueue"}
}output {stdout {codec => "rubydebug"}
}

这次我们更好了：Logstash确实可以连接并将消息转储到标准输出。

bin/logstash --config stomp.conf 
Logstash startup completed
{"message" => "Can I kick it...","@version" => "1","@timestamp" => "2015-07-22T05:42:35.016Z"
}