日日操夜夜添-日日操影院-日日草夜夜操-日日干干-精品一区二区三区波多野结衣-精品一区二区三区高清免费不卡

本文介紹了Spark結構化流媒體中的MQ源的處理方法，對大家解決問題具有一定的參考價值，需要的朋友們下面隨著小編來一起學習吧！

問題描述

我已經在Spark Structure Streaming中實現了MQ源代碼。我正在使用IBM MQ Core庫和Java 8

            <groupId>com.ibm.mq</groupId>
            <artifactId>com.ibm.mq</artifactId>
            <version>${mq.version}</version>
            <scope>provided</scope>

問題是讀取操作太慢。我正在運行我的流應用程序，有4個執行器，每個執行器4 GB內存，4個內核。但我的應用程序需要10分鐘才能從隊列中讀取1000條消息。每條消息的大小約為3kb。

從現在起，我的Spark應用程序逐個讀取消息并并行處理這些消息。

有沒有一種方法可以讓我們同時閱讀這些消息。我聽說過MQ中的并發消費者。但我找不到任何有關這方面的適當文件。

有人能用Java跟我分享一些例子嗎？

由于我的組織限制，我無法共享任何代碼。對此我深表歉意。但我可以提供任何需要的信息來提供一些幫助。

推薦答案

但我的應用程序需要10分鐘從隊列中讀取1000條消息。
每封郵件的大小約為3KB。

您是在與MQ隊列管理器相同的服務器上運行應用程序，還是在遠程服務器上運行應用程序？在循環獲取下一條消息之前，您的應用程序是否正在做一些后端工作？您進行了哪些網絡性能測試？可能您的網絡速度很慢。

下面是一個簡單的Java/MQ，它循環和檢索隊列上的所有消息。我只是將1000條3kb的消息放在隊列中并運行它。以下是輸出：

2018/06/28 17:18:21.191 MQTest12L: testReceive: successfully connected to MQWT1
2018/06/28 17:18:21.207 MQTest12L: testReceive: successfully opened TEST.Q1
2018/06/28 17:18:21.448 MQTest12L: testReceive: read 1000 messages
2018/06/28 17:18:21.448 MQTest12L: testReceive: closed: TEST.Q1
2018/06/28 17:18:21.450 MQTest12L: testReceive: disconnected from MQWT1

獲取1000條3KB郵件花費了259毫秒。因此，如果您的程序需要10分鐘才能收到1000條3kb的消息，則說明您的程序或您的網絡有問題。

下面是一個功能齊全的Java/MQ應用程序，名為MQTest12L.java，它將循環并檢索隊列中的所有消息：

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Hashtable;

import com.ibm.mq.MQException;
import com.ibm.mq.MQMessage;
import com.ibm.mq.MQGetMessageOptions;
import com.ibm.mq.MQQueue;
import com.ibm.mq.MQQueueManager;
import com.ibm.mq.constants.CMQC;

/**
 * Program Name
 *  MQTest12L
 *
 * Description
 *  This java class will connect to a remote queue manager with the 
 *  MQ setting stored in a HashTable, loop to retrieve all messages on a queue
 *  then close and disconnect.
 *
 * Sample Command Line Parameters
 *  -m MQA1 -h 127.0.0.1 -p 1414 -c TEST.CHL -q TEST.Q1 -u UserID -x Password
 *
 * @author Roger Lacroix
 */
public class MQTest12L
{
   private static final SimpleDateFormat  lOGGER_TIMESTAMP = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss.SSS");

   private Hashtable<String,String> params;
   private Hashtable<String,Object> mqht;
   private String qMgrName;
   private String outputQName;

   /**
    * The constructor
    */
   public MQTest12L()
   {
      super();
      params = new Hashtable<String,String>();
      mqht = new Hashtable<String,Object>();
   }

   /**
    * Make sure the required parameters are present.
    * @return true/false
    */
   private boolean allParamsPresent()
   {
      boolean b = params.containsKey("-h") && params.containsKey("-p") &&
                  params.containsKey("-c") && params.containsKey("-m") &&
                  params.containsKey("-q") &&
                  params.containsKey("-u") && params.containsKey("-x");
      if (b)
      {
         try
         {
            Integer.parseInt((String) params.get("-p"));
         }
         catch (NumberFormatException e)
         {
            b = false;
         }
      }

      return b;
   }

   /**
    * Extract the command-line parameters and initialize the MQ HashTable.
    * @param args
    * @throws IllegalArgumentException
    */
   private void init(String[] args) throws IllegalArgumentException
   {
      int port = 1414;
      if (args.length > 0 && (args.length % 2) == 0)
      {
         for (int i = 0; i < args.length; i += 2)
         {
            params.put(args[i], args[i + 1]);
         }
      }
      else
      {
         throw new IllegalArgumentException();
      }

      if (allParamsPresent())
      {
         qMgrName = (String) params.get("-m");
         outputQName = (String) params.get("-q");

         try
         {
            port = Integer.parseInt((String) params.get("-p"));
         }
         catch (NumberFormatException e)
         {
            port = 1414;
         }

         mqht.put(CMQC.CHANNEL_PROPERTY, params.get("-c"));
         mqht.put(CMQC.HOST_NAME_PROPERTY, params.get("-h"));
         mqht.put(CMQC.PORT_PROPERTY, new Integer(port));
         mqht.put(CMQC.USER_ID_PROPERTY, params.get("-u"));
         mqht.put(CMQC.PASSWORD_PROPERTY, params.get("-x"));

         // I don't want to see MQ exceptions at the console.
         MQException.log = null;
      }
      else
      {
         throw new IllegalArgumentException();
      }
   }

   /**
    * Connect, open queue, loop and get all messages then close queue and disconnect.
    *
    * @throws MQException
    */
   private void testReceive()
   {
      MQQueueManager qMgr = null;
      MQQueue queue = null;
      int openOptions = CMQC.MQOO_INPUT_AS_Q_DEF + CMQC.MQOO_INQUIRE + CMQC.MQOO_FAIL_IF_QUIESCING;
      MQGetMessageOptions gmo = new MQGetMessageOptions();
      gmo.options = CMQC.MQGMO_NO_WAIT + CMQC.MQGMO_FAIL_IF_QUIESCING;
      MQMessage receiveMsg = null;
      int msgCount = 0;
      boolean getMore = true;

      try
      {
         qMgr = new MQQueueManager(qMgrName, mqht);
         MQTest12L.logger("successfully connected to "+ qMgrName);

         queue = qMgr.accessQueue(outputQName, openOptions);
         MQTest12L.logger("successfully opened "+ outputQName);

         while(getMore)
         {
            receiveMsg = new MQMessage();

            try
            {
               // get the message on the queue
               queue.get(receiveMsg, gmo);
               msgCount++;

               if (CMQC.MQFMT_STRING.equals(receiveMsg.format))
               {
                  String msgStr = receiveMsg.readStringOfByteLength(receiveMsg.getMessageLength());
//                  MQTest12L.logger("["+msgCount+"] " + msgStr);
               }
               else
               {
                  byte[] b = new byte[receiveMsg.getMessageLength()];
                  receiveMsg.readFully(b);
//                  MQTest12L.logger("["+msgCount+"] " + new String(b));
               }
            }
            catch (MQException e)
            {
               if ( (e.completionCode == CMQC.MQCC_FAILED) && 
                    (e.reasonCode == CMQC.MQRC_NO_MSG_AVAILABLE) )
               {
                  // All messages read.
                  getMore = false;
                  break;
               }
               else
               {
                  MQTest12L.logger("MQException: " + e.getLocalizedMessage());
                  MQTest12L.logger("CC=" + e.completionCode + " : RC=" + e.reasonCode);
                  getMore = false;
                  break;
               }
            }
            catch (IOException e)
            {
               MQTest12L.logger("IOException:" +e.getLocalizedMessage());
            }
         }
      }
      catch (MQException e)
      {
         MQTest12L.logger("CC=" +e.completionCode + " : RC=" + e.reasonCode);
      }
      finally
      {
         MQTest12L.logger("read " + msgCount + " messages");

         try
         {
            if (queue != null)
            {
               queue.close();
               MQTest12L.logger("closed: "+ outputQName);
            }
         }
         catch (MQException e)
         {
            MQTest12L.logger("CC=" +e.completionCode + " : RC=" + e.reasonCode);
         }
         try
         {
            if (qMgr != null)
            {
               qMgr.disconnect();
               MQTest12L.logger("disconnected from "+ qMgrName);
            }
         }
         catch (MQException e)
         {
            MQTest12L.logger("CC=" +e.completionCode + " : RC=" + e.reasonCode);
         }
      }
   }

   /**
    * A simple logger method
    * @param data
    */
   public static void logger(String data)
   {
      String className = Thread.currentThread().getStackTrace()[2].getClassName();

      // Remove the package info.
      if ( (className != null) && (className.lastIndexOf('.') != -1) )
         className = className.substring(className.lastIndexOf('.')+1);

      System.out.println(lOGGER_TIMESTAMP.format(new Date())+" "+className+": "+Thread.currentThread().getStackTrace()[2].getMethodName()+": "+data);
   }

   /**
    * main line
    * @param args
    */
   public static void main(String[] args)
   {
      MQTest12L write = new MQTest12L();

      try
      {
         write.init(args);
         write.testReceive();
      }
      catch (IllegalArgumentException e)
      {
         System.err.println("Usage: java MQTest12L -m QueueManagerName -h host -p port -c channel -q QueueName -u UserID -x Password");
         System.exit(1);
      }

      System.exit(0);
   }
}

這篇關于Spark結構化流媒體中的MQ源的文章就介紹到這了，希望我們推薦的答案對大家有所幫助，