Spark 流:自定义接收器:数据源:Websphere 消息队列

Question

我正在尝试在 Spark 流中实现 WSMQ 数据源的客户接收器。我遵循了此处提供的示例。

后来我在这个 Github 存储库中遵循了示例。

我遇到了三个问题：

1：错误(此错误在程序运行一段时间后出现(

java.net.ConnectException: Connection refused
    at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method)
    at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:739)
    at org.apache.kafka.common.network.Selector.poll(Selector.java:238)
    at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:192)
    at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:191)
    at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:122)
    at java.lang.Thread.run(Thread.java:745)

2. 该程序不会从 WSMQ 中删除消息，即使我在创建会话时使用了此代码

   MQQueueSession qSession = (MQQueueSession) qCon.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
   

3. 我需要实现自定义接收器 Spark API 中解释的可靠Receiver。它说：

   要实现可靠的接收器，您必须使用store(多记录(来存储数据。这种存储风格是一个阻塞调用，只有在所有给定的记录都存储在 Spark 中后才会返回。如果接收方配置的存储级别使用复制(默认启用(，则此调用将在复制完成后返回。因此，它确保可靠地存储数据，并且接收器现在可以适当地确认源。这可确保当接收方在复制数据的过程中发生故障时不会引起任何数据 - 缓冲的数据不会被确认，因此稍后将由源重新发送。

我不知道如何处理存储(多记录(？

我无法弄清楚为什么会发生这些错误以及如何实现可靠的Receiver.

这是代码：

public class JavaConnector 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;
    String topic=null;
    Enumeration enumeration =null;
    private static MQQueueReceiver receiver = null;

    public JavaConnector(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();
                }
                catch (Exception 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() throws InterruptedException {
        System.out.print("Started receiving messages from MQ");

        try {
            JMSTextMessage receivedMessage= null;
            int cnt =0;
            //JMSTextMessage receivedMessage = (JMSTextMessage) receiver.receive(10000);
            boolean flag=false;
            while (!isStopped() && enumeration.hasMoreElements()&&cnt<50 )
            {
                receivedMessage= (JMSTextMessage) enumeration.nextElement();
                receivedMessage.acknowledge();
                String userInput = receivedMessage.getText();
                    ArrayList<String> list = new ArrayList<String>();
                    list.add(userInput);
                    Iterator<String> itr = list.iterator();
                    store(itr);
                cnt++;
            }
            /*while (!isStopped() && receivedMessage !=null)
            {
               // receivedMessage= (JMSTextMessage) enumeration.nextElement();
                String userInput = receivedMessage.getText();
                store(userInput);
        receivedMessage.acknowledge();
            }*/
            // 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)
        {      Thread.sleep(100);
            System.out.println("WRONG"+e.toString());
            e.printStackTrace();
            restart("Trying to connect again");
        }
        catch(Throwable t) {
            Thread.sleep(100);
            System.out.println("WRONG-1"+t.toString());
            // restart if there is any other error
            restart("Error receiving data", t);
        }

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

            qCon = (MQQueueConnection) conFactory.createQueueConnection();
            MQQueueSession qSession = (MQQueueSession) qCon.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
            MQQueue queue = (MQQueue) qSession.createQueue(qn);
            MQQueueBrowser browser = (MQQueueBrowser) qSession.createBrowser(queue);
            qCon.start();
            //receiver = (MQQueueReceiver) qSession.createReceiver(queue);
            enumeration= browser.getEnumeration();

        } catch (Exception e) {
            Thread.sleep(1000);
        }
    }
    @Override
    public StorageLevel storageLevel() {
        return StorageLevel.MEMORY_ONLY_2();
    }

Answer 1

终于我能够解决这个问题了。解决方案 1：蒸汽上下文尝试写入 Kafka，因为 kafka 已关闭并且它给了我 IO 错误。这对我来说太愚蠢了。:)

解决方案2：我应该使用MessageListener，QueueBrowser用于读取它实际上不消耗消息的消息。