在 Apache Flume 中获取 JMS 标头答案

【问题标题】：Fetching JMS Header in Apache Flume在 Apache Flume 中获取 JMS 标头
【发布时间】：2018-01-26 17:21:01
【问题描述】：

我正在尝试使用 Apache Flume 使用 JMS 消息 (IBM Websphere MQ) 并将数据存储到 HDFS。在阅读邮件时，我只能看到邮件的正文，而不是邮件的标题内容。

是否可以使用 Apache Flume 读取带有 header 属性的 jms 消息？

我的配置：

# Source definition
u.sources.s1.type=jms
u.sources.s1.initialContextFactory=ABC
u.sources.s1.connectionFactory=<my connection factory>
u.sources.s1.providerURL=ABC
u.sources.s1.destinationName=r1
u.sources.s1.destinationType=QUEUE
# Channel definition
u.channels.c1.type=file
u.channels.c1.capacity=10000000
u.channels.c1.checkpointDir=/checkpointdir
u.channels.c1.transactionCapacity=10000
u.channels.c1.dataDirs=/datadir
# Sink definition
u.sinks.r1.type=hdfs
u.sinks.r1.channel=c1
u.sinks.r1.hdfs.path=/message/%Y%m%d
u.sinks.r1.hdfs.filePrefix=event_
u.sinks.r1.hdfs.fileSuffix=.xml
u.sinks.r1.hdfs.fileType = DataStream
u.sinks.r1.hdfs.writeFormat=Text
u.sinks.r1.hdfs.useLocalTimeStamp=TRUE

【问题讨论】：

你指的是什么标题内容？ MQMD 还是命名属性？
@Roger 我正在寻找 MQMD 和命名属性。基本思想是存储来自 JMS MQ 的标头内容。
对于命名属性只需执行：msg.getStringProperty("name") 和对于 MQMD，请遵循此处的信息：ibm.com/support/knowledgecenter/en/SSFKSJ_7.5.0/…
Flume 1.5.0 用户指南声明与 JMS 源相关：“在所有情况下，消息中的属性都作为标头添加到 FlumeEvent。”
@JoshMc 你能提供同样的网址吗？

标签： apache hadoop ibm-mq flume flume-ng

【解决方案1】：

有很多类型的 JMS 消息，如“表 30–2 JMS 消息类型”here。

Flume DefaultJMSMessageConverter 使用 TextMessage 和 here 一样，下面给出供您参考：

...
else if(message instanceof TextMessage) {
      TextMessage textMessage = (TextMessage)message;
      event.setBody(textMessage.getText().getBytes(charset));
    } 
...

TextMessage 仅提供邮件正文。

恕我直言，您有两个选择：

如果可能，请在正文中发送消息头、头值对并按原样使用“DefaultJMSMessageConverter”。
通过编写自定义 JMSMessageConverter 并将“消息”类型转换为 javax.jms.Message 来构建您自己的“flume-jms-source.jar”，获取 JMS 标头，将它们设置在 SimpleEvent 中。

希望这能给你一些方向。

【讨论】：