Friday, 13 June 2014

Concurrency within an EJB 3

In many large apps, some processes require multiple actions to be performed. For example, user edits or creates many items on the screen, all items have to be saved one by one and cannot be done all at once. This is not mostly the case with JPA entities, but rather custom data and old way of doing things. This can be the case for example when inserts are done to different tables or databases and on top of that result from one insert is used for the next one. Unfortunately there is tons of interesting data models with which developers and users have to deal with. In order to relieve already overstressed users from waiting on submit for a set of newly created or edited items, one may decide to execute them in parallel and when all is finished, display the results to user. In the context of en EJB 3.1, this is what the spec says:

"There is no need for any restrictions against concurrent client access to stateless session beans because the container routes each request to a different instance of the stateless session bean class..."

What if, I have a list of date that I want to persist concurrently, each in a thread, wait until all is finished and proceed? Assuming I am not using @Asynchronous option, which is only available in 3.1. So I tried to use an executor service:

EJB which is supposed to be called from different threads:

1
2
3
4
5
6
@Local
public interface TestConcurrentEJBAccess {

 public void test();
 
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
@Stateless
public class TestConcurrentEJBAccessBean implements TestConcurrentEJBAccess {

 @Override
 public void test() {
  System.out.println("Inside the TestConcurrentEJBAccessBean: " + this);
  
 }

}

Following is an actual caller implementation:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
 
@Stateless
public class ManageServiceBean implements ManageService {
 
 @EJB
 private TestConcurrentEJBAccess testConcurrentEJBAccess;
 ...

 @Override
 public int saveTable(List<Val> lookup, final String user) {
  ...
  List<Callable<Integer>> callables = new ArrayList<Callable<Integer>>();
  ExecutorService es = null;
  try {
   final InitialContext sctx = new InitialContext();

   for (final LookupVal lookupValue : changed) {

    Callable<Integer> c = new Callable<Integer>() {
     @Override
     public Integer call() throws Exception {
      // execute test
      testConcurrentEJBAccess.test();
      
     }
    };

    callables.add(c);
   }
   
   es = Executors.newCachedThreadPool();
   List<Future<Integer>> results = es.invokeAll(callables);

   System.out.println(System.currentTimeMillis() - start);

   for (Future<Integer> future : results) {
    totalResult += future.get();
   }
   
   // Next is exception handling other not interesting stuff
   ...
 }
} 
     

Here is the console output:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
Inside the TestConcurrentEJBAccessBean: com.test.conc.aplf.ejb.spb.TestConcurren
tEJBAccessBean_6ali4g_Impl@19f72db
Inside the TestConcurrentEJBAccessBean: com.test.conc.aplf.ejb.spb.TestConcurren
tEJBAccessBean_6ali4g_Impl@19f72db
Inside the TestConcurrentEJBAccessBean: com.test.conc.aplf.ejb.spb.TestConcurren
tEJBAccessBean_6ali4g_Impl@19f72db
Inside the TestConcurrentEJBAccessBean: com.test.conc.aplf.ejb.spb.TestConcurren
tEJBAccessBean_6ali4g_Impl@19f72db
Inside the TestConcurrentEJBAccessBean: com.test.conc.aplf.ejb.spb.TestConcurren
tEJBAccessBean_6ali4g_Impl@19f72db

Same EJB instance is being invoked, just from separate threads! But that is not what i wanted. I wanted separate for each "client". Each invocation does however run in its own transaction, as seen from following modifications:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
import javax.annotation.Resource;
import javax.ejb.Stateless;
import javax.transaction.TransactionSynchronizationRegistry;

import weblogic.transaction.internal.ServerTransactionImpl;

@Stateless
public class TestConcurrentEJBAccessBean implements TestConcurrentEJBAccess {

 @Resource
 private TransactionSynchronizationRegistry tsr;
 
 @Override
 public void test() {
  System.out.println("Inside the TestConcurrentEJBAccessBean: " + this + 
       " transaction: " + ((ServerTransactionImpl)tsr.getTransactionKey()).getXid());
  
 }

}

and output:

1
2
3
4
5
6
7
8
Inside the TestConcurrentEJBAccessBean: com.test.conc.aplf.ejb.spb.TestConcurren
tEJBAccessBean_6ali4g_Impl@1e20bd1 transaction: BEA1-025A2FE57008E8E9B79B
Inside the TestConcurrentEJBAccessBean: com.test.conc.aplf.ejb.spb.TestConcurren
tEJBAccessBean_6ali4g_Impl@115b3a transaction: BEA1-02592FE57008E8E9B79B
Inside the TestConcurrentEJBAccessBean: com.test.conc.aplf.ejb.spb.TestConcurren
tEJBAccessBean_6ali4g_Impl@175d190 transaction: BEA1-025B2FE57008E8E9B79B
Inside the TestConcurrentEJBAccessBean: com.test.conc.aplf.ejb.spb.TestConcurren
tEJBAccessBean_6ali4g_Impl@175d190 transaction: BEA1-025C2FE57008E8E9B79B

Transaction ID is different in all cases. So this is ok. Also from the output there is 3 separate instances and not one, like I thought at first. Lets however try to get a separate instance per callable. For that we modify our ManageServiceBean on how we invoke TestConcurrentEJBAccess:

1
2
3
4
5
6
// execute test
//testConcurrentEJBAccess.test();
        
((TestConcurrentEJBAccess) sctx
  .lookup("java:comp/env/ejb/TestConcurrentEJBAccess"))
  .test();

Now the output:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
Inside the TestConcurrentEJBAccessBean: com.test.conc.aplf.ejb.spb.testconcurren
tservice_g1cl98_Impl@a6ed65 transaction: BEA1-03AC2FE57008E8E9B79B
Inside the TestConcurrentEJBAccessBean: com.test.conc.aplf.ejb.spb.testconcurren
tservice_g1cl98_Impl@a6ed65 transaction: BEA1-03B22FE57008E8E9B79B
Inside the TestConcurrentEJBAccessBean: com.test.conc.aplf.ejb.spb.testconcurren
tservice_g1cl98_Impl@a6ed65 transaction: BEA1-03B42FE57008E8E9B79B
Inside the TestConcurrentEJBAccessBean: com.test.conc.aplf.ejb.spb.testconcurren
tservice_g1cl98_Impl@a6ed65 transaction: BEA1-03B62FE57008E8E9B79B
Inside the TestConcurrentEJBAccessBean: com.test.conc.aplf.ejb.spb.testconcurren
tservice_g1cl98_Impl@a6ed65 transaction: BEA1-03B82FE57008E8E9B79B
Inside the TestConcurrentEJBAccessBean: com.test.conc.aplf.ejb.spb.testconcurren
tservice_g1cl98_Impl@a6ed65 transaction: BEA1-03BB2FE57008E8E9B79B
Inside the TestConcurrentEJBAccessBean: com.test.conc.aplf.ejb.spb.testconcurren
tservice_g1cl98_Impl@a6ed65 transaction: BEA1-03BF2FE57008E8E9B79B
Inside the TestConcurrentEJBAccessBean: com.test.conc.aplf.ejb.spb.testconcurren
tservice_g1cl98_Impl@a6ed65 transaction: BEA1-03C02FE57008E8E9B79B
Inside the TestConcurrentEJBAccessBean: com.test.conc.aplf.ejb.spb.testconcurren
tservice_g1cl98_Impl@a6ed65 transaction: BEA1-03C12FE57008E8E9B79B
Inside the TestConcurrentEJBAccessBean: com.test.conc.aplf.ejb.spb.testconcurren
tservice_g1cl98_Impl@a6ed65 transaction: BEA1-03C22FE57008E8E9B79B
Inside the TestConcurrentEJBAccessBean: com.test.conc.aplf.ejb.spb.testconcurren
tservice_g1cl98_Impl@a6ed65 transaction: BEA1-03C32FE57008E8E9B79B
Inside the TestConcurrentEJBAccessBean: com.test.conc.aplf.ejb.spb.testconcurren
tservice_g1cl98_Impl@a6ed65 transaction: BEA1-03C42FE57008E8E9B79B
Inside the TestConcurrentEJBAccessBean: com.test.conc.aplf.ejb.spb.testconcurren
tservice_g1cl98_Impl@a6ed65 transaction: BEA1-03C52FE57008E8E9B79B

The instance is the same! In any case there is no control of which instance of a stateless EJB, the container will provide...

In any case it doesn't matter same instance or not. What matters is if concurrency inside an EJB is justifying enough or not. This needs to be measured and calculated, since there may be a single point of synchronization for running threads, waiting for a lock in which will not provide any benefit for parralelizing an algorithm.

Wednesday, 11 June 2014

Прямое соединение с MQ не используя JMS

Часто в крупных корпорациях в качестве продукта системы сообщений, используют MQ. Это IBM-овский софт заслуживший свою репутацию за счет надежности, ну и потому что был одним из первых. В основном в Java EE/J2EE его интегрировали через JMS, это стандартный ход который поощряется, тем не менее возможно использовать этого провайдера на прямую, так сказать без прокладки JMS.      

Ниже представлен EJB 3 @Stateless бин выполняющий роль связного. Это совсем не обязательно должен быть EJB 3 бином, результата можно добиться и простым объектом, просто имея при себе Java EE сервер, поддерживающий специфику бинов с накопителями и транзакциями, этим лучше воспользоваться, так как при большом потоке, а программа использует этот коннектор в качестве связного по практически всем вопросам кредитных карт банка, в общем кол-во запросов колоссальное. По скорости против простых объектов, EJB конечно остался победителем. Можно это выполнить и в статическом методе, но руководить бинами проще и в случае ошибок и с точки зрения о возможных модификациях в будущем, да и мониторить их проще.



  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
package com.plm.MAINFRxml.generic.processing.xml;

import java.util.Properties;

import javax.naming.Context;
import javax.naming.NamingException;
import javax.naming.directory.InitialDirContext;

import com.plm.MAINFRxml.exception.MQAdapterException;
import com.plm.MAINFRxml.service.ApplicationService;
import com.plm.MAINFRxml.service.ResourceRetriever;
import com.ibm.mq.MQC;
import com.ibm.mq.MQEnvironment;
import com.ibm.mq.MQException;
import com.ibm.mq.MQGetMessageOptions;
import com.ibm.mq.MQMessage;
import com.ibm.mq.MQPutMessageOptions;
import com.ibm.mq.MQQueue;
import com.ibm.mq.MQQueueManager;

/**
 *
 * Sends message to MAINFR.
 * @author Emil Iakoupov
 *
 */
@Stateless
public class MAINFRConnectorEJB implements MQConnectionEnv{

    /**
  * Serial id.
  */
 private static final long serialVersionUID = 1L;

 private static final String INQUIRE_LOCAL_QUEUE = "inquire.local.queue";
 private static final String INQUIRE_RESPOND_LOCAL_QUEUE = "inquire.respond.local.queue";
 private static final String MAINTAIN_LOCAL_QUEUE = "maintain.local.queue";
 private static final String MAINTAIN_RESPOND_LOCAL_QUEUE = "maintain.respond.local.queue";
 private static final String EVT_LOCAL_QUEUE = "event.local.queue";
 private static final String EVT_RESPOND_LOCAL_QUEUE = "event.respond.local.queue";
 
 private String qManager = null;
 private String hostname = null;
 private String channel = null;
 private int port = 0;
 private String userID = null;
 private String password = null;

 private static final String TIMEOUT = "timeout";
 private static final String EXPIRE = "expire";

 private InitialDirContext ctx;

 private Properties envpr;


    /**
     * Sends message to MAINFR and receives response back.
     *
     * @param msg MAINFR format string message
     * @return response message string from MAINFR will be returned
     * @throws MQAdapterException
     */
    public String processMessage(String msg, String requestType)
    throws MQAdapterException {
     String response = null;

     response = sendMessage(msg, requestType);


  return response;
    }

    /**
     * Reads bean environment from the initial context.
     * @throws MQAdapterException
     */
    private void readEnv() {
      
     try {

          envpr = ResourceRetriever.RETRIEVER.getProperties("Connector");
          
          qManager = envpr.getProperty("queue.manager").trim();
         hostname = envpr.getProperty("host.name").trim();
         channel = envpr.getProperty("channel").trim();
         port = Integer.parseInt(envpr.getProperty("port").trim());
         userID = envpr.getProperty("user.id").trim();
         password = envpr.getProperty("user.password").trim();

        } catch(Exception ne) {
            ne.printStackTrace();
        } 


    }

    /**
     * Send message.
     * @throws MQAdapterException
     */
    public String sendMessage(String pack, String requestType)
    throws MQAdapterException {

     String response = null;

     String qName = null;
     String responseQName = null;
        MQQueueManager qMgr = null;
        MQQueue queue = null;
        MQQueue getQueue = null;
     try {
 
   qMgr = getConnection(envpr);
   qName = getSndQ(requestType);
 
   /*int openOptions = 17;*/
   
   queue = qMgr.accessQueue(qName, MQC.MQOO_OUTPUT );
   
   MQMessage msg = new MQMessage();
   
            //msg.correlationId = hexToByte(id);
            //msg.messageId = id.getBytes();
            
            responseQName = getRcvQ(requestType);
            getQueue = qMgr.accessQueue(responseQName, MQC.MQOO_INPUT_SHARED | MQC.MQOO_FAIL_IF_QUIESCING);
            
            msg.replyToQueueName = getQueue.name;
   //putMessage.replyToQueueManagerName = mQueueManagerName;
            msg.replyToQueueManagerName = "CIMSD01";
            //msg.messageType = MQC.MQMT_REQUEST;
            msg.format = MQC.MQFMT_STRING;
            msg.encoding = MQC.MQENC_NATIVE;
            msg.characterSet = MQC.MQCCSI_DEFAULT; // 37
            
            msg.writeString(pack);
            
            MQPutMessageOptions pmo = new MQPutMessageOptions();
            pmo.options = MQC.MQPMO_FAIL_IF_QUIESCING;
            
            queue.put(msg, pmo);
            
            MQMessage getMessage = new MQMessage();
   //Set the get message MQMD parameters
   //getMessage.messageId = putMessage.messageId;
   getMessage.correlationId = msg.messageId;
   getMessage.encoding = MQC.MQENC_NATIVE;
   getMessage.characterSet = MQC.MQCCSI_DEFAULT;
   
   MQGetMessageOptions mGetMessageOptions = new MQGetMessageOptions();
   mGetMessageOptions.waitInterval = Integer.valueOf(envpr.getProperty(TIMEOUT).trim());
   // configured milliseconds
   mGetMessageOptions.matchOptions = MQC.MQMO_MATCH_CORREL_ID;
   
   mGetMessageOptions.options = 
    MQC.MQGMO_WAIT | 
    MQC.MQGMO_CONVERT | 
    MQC.MQGMO_FAIL_IF_QUIESCING;
   
   getQueue.get(getMessage, mGetMessageOptions);
   
   response = getMessage.readString(getMessage.getMessageLength());

  } catch (MQException e) {
   // throw checked exception here so its propagated to the client
   e.printStackTrace();
   throw new MQAdapterException(e);
         
  } catch (NamingException e) {
   e.printStackTrace();
  } catch (Throwable e) {
   e.printStackTrace();
  } finally {
   try {
    if (queue != null)
     queue.close();
    
    if (getQueue != null)
     getQueue.close();
    
    if (qMgr != null)
     qMgr.disconnect();
     
   } catch (MQException e) {
    e.printStackTrace();
   }
  }
  
  return response;


    }

    /**
     * Gets MQQueueManager.
     * @return
     */
 private MQQueueManager getConnection(Properties props) {
  
  MQQueueManager queueManager = null;
  try
  {
   // specify MQI client to MQSeries environment
   MQEnvironment.hostname = hostname;
   MQEnvironment.channel = channel;
   if (port > 0)
   {
    MQEnvironment.port = port;
   }
   if (userID != null)
   {
    MQEnvironment.userID = userID;
   }
   if (password != null)
   {
    MQEnvironment.password = password;
   }
   MQEnvironment.properties.put(MQC.TRANSPORT_PROPERTY, MQC.TRANSPORT_MQSERIES);

   queueManager = new MQQueueManager(qManager);
   
  } catch (MQException ex) {
   ex.printStackTrace();
  }

 
  return queueManager;
 }

    private String getSndQ(String requestType) throws NamingException {

     String name = null;

     if (requestType.equalsIgnoreCase(ApplicationService.INQUIRY)) {
      name =  envpr.getProperty(INQUIRE_LOCAL_QUEUE);

     } else if (requestType.equalsIgnoreCase(ApplicationService.MAINTENANCE)) {
      name =  envpr.getProperty(MAINTAIN_LOCAL_QUEUE);

     } else if (requestType.equalsIgnoreCase(ApplicationService.EVENT)) {
      name =  envpr.getProperty(EVT_LOCAL_QUEUE);
     }



     return name;
    }


    private String getRcvQ(String requestType) throws NamingException {

     String name = null;
 

     if (requestType.equalsIgnoreCase(ApplicationService.INQUIRY)) {
      name =  envpr.getProperty(INQUIRE_RESPOND_LOCAL_QUEUE);

     } else if (requestType.equalsIgnoreCase(ApplicationService.MAINTENANCE)) {
      name =  envpr.getProperty(MAINTAIN_RESPOND_LOCAL_QUEUE);

     } else if (requestType.equalsIgnoreCase(ApplicationService.EVENT)) {
      name =  envpr.getProperty(EVT_RESPOND_LOCAL_QUEUE);
     }

     return name;
    }

}

Вот проперти файл:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
inquire.local.queue=BS3V.SYS.A1680.INQ
inquire.respond.local.queue=LIM1.LIM.A1680INQ7.RS01

maintain.local.queue=BS3V.SYS.A1680.MAINT
maintain.respond.local.queue=LIM1.LIM.A1680MNT7.RS01

event.local.queue=BS3V.SYS.A1680.EVT
event.respond.local.queue=LIM1.LIM.A1680EVT7.RS01

# timeout
timeout=10000

# expiration
expire=12000

queue.manager=LIM1
connection.type=CLIENT
host.name=192.168.35.197 
channel=LIM1.SRVCON.CHA01
port=1414
user.id=
user.password=

Значит так. С JMS-ом, такого же типа бин, операции посылки/отправки проводил медленнее, что особенно было заметно при большом потоке, не говоря о том что JMS по умолчанию сохраняет сообщения (это очень тормозит процесс при огромном потоке), но это можно отключить. В данном случае использовался Weblogic 10.3. 

Direct MQ communication in Java EE without JMS

Witrhin Java EE environment, MQ is often used for communication with legacy systems, due to its reputation for being reliable. This is usially configured through JMS, this is however is an implementation without using JMS, which turned out to be simpler and easier to implement.

Following is an EJB 3 stateless bean, which acts as a sender and receiver of MQ message:


  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
package com.plm.MAINFRxml.generic.processing.xml;

import java.util.Properties;

import javax.naming.Context;
import javax.naming.NamingException;
import javax.naming.directory.InitialDirContext;

import com.plm.MAINFRxml.exception.MQAdapterException;
import com.plm.MAINFRxml.service.ApplicationService;
import com.plm.MAINFRxml.service.ResourceRetriever;
import com.ibm.mq.MQC;
import com.ibm.mq.MQEnvironment;
import com.ibm.mq.MQException;
import com.ibm.mq.MQGetMessageOptions;
import com.ibm.mq.MQMessage;
import com.ibm.mq.MQPutMessageOptions;
import com.ibm.mq.MQQueue;
import com.ibm.mq.MQQueueManager;

/**
 *
 * Sends message to MAINFR.
 * @author Emil Iakoupov
 *
 */
@Stateless
public class MAINFRConnectorEJB implements MQConnectionEnv{

    /**
  * Serial id.
  */
 private static final long serialVersionUID = 1L;

 private static final String INQUIRE_LOCAL_QUEUE = "inquire.local.queue";
 private static final String INQUIRE_RESPOND_LOCAL_QUEUE = "inquire.respond.local.queue";
 private static final String MAINTAIN_LOCAL_QUEUE = "maintain.local.queue";
 private static final String MAINTAIN_RESPOND_LOCAL_QUEUE = "maintain.respond.local.queue";
 private static final String EVT_LOCAL_QUEUE = "event.local.queue";
 private static final String EVT_RESPOND_LOCAL_QUEUE = "event.respond.local.queue";
 
 private String qManager = null;
 private String hostname = null;
 private String channel = null;
 private int port = 0;
 private String userID = null;
 private String password = null;

 private static final String TIMEOUT = "timeout";
 private static final String EXPIRE = "expire";

 private InitialDirContext ctx;

 private Properties envpr;


    /**
     * Sends message to MAINFR and receives response back.
     *
     * @param msg MAINFR format string message
     * @return response message string from MAINFR will be returned
     * @throws MQAdapterException
     */
    public String processMessage(String msg, String requestType)
    throws MQAdapterException {
     String response = null;

     response = sendMessage(msg, requestType);


  return response;
    }

    /**
     * Reads bean environment from the initial context.
     * @throws MQAdapterException
     */
    private void readEnv() {
      
     try {

          envpr = ResourceRetriever.RETRIEVER.getProperties("Connector");
          
          qManager = envpr.getProperty("queue.manager").trim();
         hostname = envpr.getProperty("host.name").trim();
         channel = envpr.getProperty("channel").trim();
         port = Integer.parseInt(envpr.getProperty("port").trim());
         userID = envpr.getProperty("user.id").trim();
         password = envpr.getProperty("user.password").trim();

        } catch(Exception ne) {
            ne.printStackTrace();
        } 


    }

    /**
     * Send message.
     * @throws MQAdapterException
     */
    public String sendMessage(String pack, String requestType)
    throws MQAdapterException {

     String response = null;

     String qName = null;
     String responseQName = null;
        MQQueueManager qMgr = null;
        MQQueue queue = null;
        MQQueue getQueue = null;
     try {
 
   qMgr = getConnection(envpr);
   qName = getSndQ(requestType);
 
   /*int openOptions = 17;*/
   
   queue = qMgr.accessQueue(qName, MQC.MQOO_OUTPUT );
   
   MQMessage msg = new MQMessage();
   
            //msg.correlationId = hexToByte(id);
            //msg.messageId = id.getBytes();
            
            responseQName = getRcvQ(requestType);
            getQueue = qMgr.accessQueue(responseQName, MQC.MQOO_INPUT_SHARED | MQC.MQOO_FAIL_IF_QUIESCING);
            
            msg.replyToQueueName = getQueue.name;
   //putMessage.replyToQueueManagerName = mQueueManagerName;
            msg.replyToQueueManagerName = "CIMSD01";
            //msg.messageType = MQC.MQMT_REQUEST;
            msg.format = MQC.MQFMT_STRING;
            msg.encoding = MQC.MQENC_NATIVE;
            msg.characterSet = MQC.MQCCSI_DEFAULT; // 37
            
            msg.writeString(pack);
            
            MQPutMessageOptions pmo = new MQPutMessageOptions();
            pmo.options = MQC.MQPMO_FAIL_IF_QUIESCING;
            
            queue.put(msg, pmo);
            
            MQMessage getMessage = new MQMessage();
   //Set the get message MQMD parameters
   //getMessage.messageId = putMessage.messageId;
   getMessage.correlationId = msg.messageId;
   getMessage.encoding = MQC.MQENC_NATIVE;
   getMessage.characterSet = MQC.MQCCSI_DEFAULT;
   
   MQGetMessageOptions mGetMessageOptions = new MQGetMessageOptions();
   mGetMessageOptions.waitInterval = Integer.valueOf(envpr.getProperty(TIMEOUT).trim());
   // configured milliseconds
   mGetMessageOptions.matchOptions = MQC.MQMO_MATCH_CORREL_ID;
   
   mGetMessageOptions.options = 
    MQC.MQGMO_WAIT | 
    MQC.MQGMO_CONVERT | 
    MQC.MQGMO_FAIL_IF_QUIESCING;
   
   getQueue.get(getMessage, mGetMessageOptions);
   
   response = getMessage.readString(getMessage.getMessageLength());

  } catch (MQException e) {
   // throw checked exception here so its propagated to the client
   e.printStackTrace();
   throw new MQAdapterException(e);
         
  } catch (NamingException e) {
   e.printStackTrace();
  } catch (Throwable e) {
   e.printStackTrace();
  } finally {
   try {
    if (queue != null)
     queue.close();
    
    if (getQueue != null)
     getQueue.close();
    
    if (qMgr != null)
     qMgr.disconnect();
     
   } catch (MQException e) {
    e.printStackTrace();
   }
  }
  
  return response;


    }

    /**
     * Gets MQQueueManager.
     * @return
     */
 private MQQueueManager getConnection(Properties props) {
  
  MQQueueManager queueManager = null;
  try
  {
   // specify MQI client to MQSeries environment
   MQEnvironment.hostname = hostname;
   MQEnvironment.channel = channel;
   if (port > 0)
   {
    MQEnvironment.port = port;
   }
   if (userID != null)
   {
    MQEnvironment.userID = userID;
   }
   if (password != null)
   {
    MQEnvironment.password = password;
   }
   MQEnvironment.properties.put(MQC.TRANSPORT_PROPERTY, MQC.TRANSPORT_MQSERIES);

   queueManager = new MQQueueManager(qManager);
   
  } catch (MQException ex) {
   ex.printStackTrace();
  }

 
  return queueManager;
 }

    private String getSndQ(String requestType) throws NamingException {

     String name = null;

     if (requestType.equalsIgnoreCase(ApplicationService.INQUIRY)) {
      name =  envpr.getProperty(INQUIRE_LOCAL_QUEUE);

     } else if (requestType.equalsIgnoreCase(ApplicationService.MAINTENANCE)) {
      name =  envpr.getProperty(MAINTAIN_LOCAL_QUEUE);

     } else if (requestType.equalsIgnoreCase(ApplicationService.EVENT)) {
      name =  envpr.getProperty(EVT_LOCAL_QUEUE);
     }



     return name;
    }


    private String getRcvQ(String requestType) throws NamingException {

     String name = null;
 

     if (requestType.equalsIgnoreCase(ApplicationService.INQUIRY)) {
      name =  envpr.getProperty(INQUIRE_RESPOND_LOCAL_QUEUE);

     } else if (requestType.equalsIgnoreCase(ApplicationService.MAINTENANCE)) {
      name =  envpr.getProperty(MAINTAIN_RESPOND_LOCAL_QUEUE);

     } else if (requestType.equalsIgnoreCase(ApplicationService.EVENT)) {
      name =  envpr.getProperty(EVT_RESPOND_LOCAL_QUEUE);
     }

     return name;
    }

}

Now here is the properties file that has configurations:
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
inquire.local.queue=BS3V.SYS.A1680.INQ
inquire.respond.local.queue=LIM1.LIM.A1680INQ7.RS01

maintain.local.queue=BS3V.SYS.A1680.MAINT
maintain.respond.local.queue=LIM1.LIM.A1680MNT7.RS01

event.local.queue=BS3V.SYS.A1680.EVT
event.respond.local.queue=LIM1.LIM.A1680EVT7.RS01

# timeout
timeout=10000

# expiration
expire=12000

queue.manager=LIM1
connection.type=CLIENT
host.name=192.168.35.197 
channel=LIM1.SRVCON.CHA01
port=1414
user.id=
user.password=

As you may notice, the correlation id is used to grab the message from the queue. During the testing in which I compared this to MQ over JMS, on Weblogic 10.3, this was faster. Code has lots of commented out lines, i haven't got a chance to format it proprely either :(. The packet here is using 3 different types of queues, depending on a message. Again this is just to show how to do a direct MQ communication with Java EE at hand.