Websphere MQ 作为 Apache Spark Streaming 的数据源

Question

我正在研究将 Websphere MQ 作为 spark-streaming 的数据源的可能性，因为在我们的一个用例中需要它。 我知道MQTT 是支持来自 MQ 数据结构的通信的协议，但由于我是触发流式传输的新手，我需要一些相同的工作示例。 有没有人尝试将 MQ 与火花流连接。请为此设计最佳方式。

Answer 1

相信你可以使用JMS来连接Websphere MQ，而Apache Camel可以用来连接Websphere MQ。您可以像这样创建自定义接收器（请注意，此模式也可以在没有 JMS 的情况下使用）：

class JMSReceiver(topicName: String, cf: String, jndiProviderURL: String)
  extends Receiver[String](StorageLevel.MEMORY_AND_DISK_SER) with Serializable  {
  //Transient as this will get passed to the Workers from the Driver
  @transient
  var camelContextOption: Option[DefaultCamelContext] = None

  def onStart() = {
    camelContextOption = Some(new DefaultCamelContext())
    val camelContext = camelContextOption.get
    val env = new Properties()
    env.setProperty("java.naming.factory.initial", "???")
    env.setProperty("java.naming.provider.url", jndiProviderURL)
    env.setProperty("com.webmethods.jms.clientIDSharing", "true")
    val namingContext = new InitialContext(env);  //using the properties file to create context

    //Lookup Connection Factory
    val connectionFactory = namingContext.lookup(cf).asInstanceOf[javax.jms.ConnectionFactory]
    camelContext.addComponent("jms", JmsComponent.jmsComponentAutoAcknowledge(connectionFactory))

    val builder = new RouteBuilder() {
        def configure() = {
          from(s"jms://topic:$topicName?jmsMessageType=Object&clientId=$clientId&durableSubscriptionName=${topicName}_SparkDurable&maxConcurrentConsumers=10")
            .process(new Processor() {
            def process(exchange: Exchange) = {
              exchange.getIn.getBody match {
                case s: String => store(s)
              }
            }
          })
        }
      }
    }
    builders.foreach(camelContext.addRoutes)
    camelContext.start()
  }

  def onStop() = if(camelContextOption.isDefined) camelContextOption.get.stop()
}

然后您可以像这样创建事件的 DStream：

val myDStream = ssc.receiverStream(new JMSReceiver("MyTopic", "MyContextFactory", "MyJNDI"))

Answer 2

所以，我在这里发布了用于连接 Websphere MQ 并读取数据的 CustomMQReceiver 的工作代码：

public class CustomMQReciever extends Receiver<String> { String host = null;
int port = -1;
String qm=null;
String qn=null;
String channel=null;
transient Gson gson=new Gson();
transient MQQueueConnection qCon= null;

Enumeration enumeration =null;

public CustomMQReciever(String host , int port, String qm, String channel, String qn) {
    super(StorageLevel.MEMORY_ONLY_2());
    this.host = host;
    this.port = port;
    this.qm=qm;
    this.qn=qn;
    this.channel=channel;

}

public void onStart() {
    // Start the thread that receives data over a connection
    new Thread()  {
        @Override public void run() {
            try {
                initConnection();
                receive();
            }
            catch (JMSException ex)
            {
                ex.printStackTrace();
            }
        }
    }.start();
}
public void onStop() {
    // There is nothing much to do as the thread calling receive()
    // is designed to stop by itself isStopped() returns false
}

 /** Create a MQ connection and receive data until receiver is stopped */
private void receive() {
  System.out.print("Started receiving messages from MQ");

    try {

    JMSMessage receivedMessage= null;

        while (!isStopped() && enumeration.hasMoreElements() )
        {

            receivedMessage= (JMSMessage) enumeration.nextElement();
            String userInput = convertStreamToString(receivedMessage);
            //System.out.println("Received data :'" + userInput + "'");
            store(userInput);
        }

        // Restart in an attempt to connect again when server is active again
        //restart("Trying to connect again");

        stop("No More Messages To read !");
        qCon.close();
        System.out.println("Queue Connection is Closed");

    }
    catch(Exception e)
    {
        e.printStackTrace();
        restart("Trying to connect again");
    }
    catch(Throwable t) {
        // restart if there is any other error
        restart("Error receiving data", t);
    }
    }

  public void initConnection() throws JMSException
{
    MQQueueConnectionFactory conFactory= new MQQueueConnectionFactory();
    conFactory.setHostName(host);
    conFactory.setPort(port);
    conFactory.setTransportType(JMSC.MQJMS_TP_CLIENT_MQ_TCPIP);
    conFactory.setQueueManager(qm);
    conFactory.setChannel(channel);


    qCon= (MQQueueConnection) conFactory.createQueueConnection();
    MQQueueSession qSession=(MQQueueSession) qCon.createQueueSession(false, 1);
    MQQueue queue=(MQQueue) qSession.createQueue(qn);
    MQQueueBrowser browser = (MQQueueBrowser) qSession.createBrowser(queue);
    qCon.start();

    enumeration= browser.getEnumeration();
   }

 @Override
public StorageLevel storageLevel() {
    return StorageLevel.MEMORY_ONLY_2();
}
}