Buffer
Email
Print

Ravish Mody

To configure a Uniform Distributed Queue you can follow the steps provided in the post Steps to Create Uniform Distributed Queue (UDQ) on Weblogic Server.

In this post I would be using a stand-alone java code for producing messages as a Producer/Clint and  for consuming the messages as a Consumer/Listener. By which we can make sure that if something goes wrong then we can narrow down the issue if its an issue with WLS or the code, as many times their an issue with the Producer or a Consumers.

.

Most Common Problems with UDQ’s

  • Un-even load balancing
  • No load balancing

The most of the time people come across the issue that the UDQ’s not able to load balance the messages evenly when they are using uniform distributed queues, or that the messages are only going to one server and not to the other servers.

Tips to solve these issues

  • Make sure the Server Affinity Enabled parameter should be un-checked (disable) which under the [ Connection factory –> Configuration (tab) –> Load Balance (sub-tab)]
  • Disable the Server Affinity Enabled parameter for the connection factory whish is been used by your UDQ.
  • All the managed servers are in the same cluster.
  • If the managed serves are in different boxes make sure the listing address is given correctly which is under [ Machine –> Configuration (tab) –> Node Manager (sub-tab) ]
  • Test if you are able to PING the servers on different boxes and make sure that there is no network issues and you are able to communicate with the servers.

Steps of test the UDQ

  1. Create a Directory somewhere in your file system like: “D:OracleJMSQueue” to write the QueueSend.java and QueueReceive.java programs.
  2. In QueueSend.java copy the below command 
    
import java.io.*;
import java.util.Hashtable;
import javax.jms.JMSException;
import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.QueueSender;
import javax.jms.QueueSession;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;

public class QueueSend
{
public final static String JNDI_FACTORY="weblogic.jndi.WLInitialContextFactory";

// ******* Give the CONNECTION FACTORY JNDI name which is UDQ_CF ********
public final static String JMS_FACTORY="UDQ_CF";

// ******* Give the UNIFORM DISTRIBUTED QUEUE JNDI name which is UDQ   ********
public final static String QUEUE="UDQ";

private QueueConnectionFactory qconFactory;
private QueueConnection qcon;
private QueueSession qsession;
private QueueSender qsender;
private Queue queue;
private TextMessage msg;

public void init(Context ctx, String queueName)
throws NamingException, JMSException
{
qconFactory = (QueueConnectionFactory) ctx.lookup(JMS_FACTORY);
qcon = qconFactory.createQueueConnection();
qsession = qcon.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
queue = (Queue) ctx.lookup(queueName);
qsender = qsession.createSender(queue);
msg = qsession.createTextMessage();
qcon.start();
}

public void send(String message) throws JMSException {
msg.setText(message);
qsender.send(msg);
}

public void close() throws JMSException {
qsender.close();
qsession.close();
qcon.close();
}

public static void main(String[] args) throws Exception {
if (args.length != 1) {
System.out.println("Usage: java QueueSend WebLogicURL");
return;
}
InitialContext ic = getInitialContext(args[0]);
QueueSend qs = new QueueSend();
qs.init(ic, QUEUE);
readAndSend(qs);
qs.close();
}

private static void readAndSend(QueueSend qs) throws IOException, JMSException
{
String line="Test Message Body with counter = ";
BufferedReader br=new BufferedReader(new InputStreamReader(System.in));
boolean readFlag=true;
System.out.println("ntStart Sending Messages (Enter QUIT to Stop):n");
while(readFlag)
{
System.out.print("<Msg_Sender> ");
String msg=br.readLine();
if(msg.equals("QUIT") || msg.equals("quit"))
{
qs.send(msg);
System.exit(0);
}
qs.send(msg);
System.out.println();
}
br.close();
}

private static InitialContext getInitialContext(String url) throws NamingException
{
Hashtable env = new Hashtable();
env.put(Context.INITIAL_CONTEXT_FACTORY, JNDI_FACTORY);
env.put(Context.PROVIDER_URL, url);
return new InitialContext(env);
}
}
  3. In QueueReceive.java copy the below command 
    import java.util.Hashtable;
import javax.jms.*;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;

public class QueueReceive implements MessageListener
{
public final static String JNDI_FACTORY="weblogic.jndi.WLInitialContextFactory";

// ******* Give the CONNECTION FACTORY JNDI name which is UDQ_CF ********
public final static String JMS_FACTORY="UDQ_CF";

// ******* Give the UNIFORM DISTRIBUTED QUEUE JNDI name which is UDQ   ********
public final static String QUEUE="UDQ";
private QueueConnectionFactory qconFactory;
private QueueConnection qcon;
private QueueSession qsession;
private QueueReceiver qreceiver;
private Queue queue;
private boolean quit = false;

public void onMessage(Message msg)
{
try {
String msgText;
if (msg instanceof TextMessage)
{
msgText = ((TextMessage)msg).getText();
}
else
{
msgText = msg.toString();
}
System.out.println("nt<Msg_Receiver> "+ msgText );
if (msgText.equalsIgnoreCase("quit"))
{
synchronized(this)
{
quit = true;
this.notifyAll(); // Notify main thread to quit
}
}
}
catch (JMSException jmse)
{
jmse.printStackTrace();
}
}
public void init(Context ctx, String queueName) throws NamingException, JMSException
{
qconFactory = (QueueConnectionFactory) ctx.lookup(JMS_FACTORY);
qcon = qconFactory.createQueueConnection();
qsession = qcon.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
queue = (Queue) ctx.lookup(queueName);
qreceiver = qsession.createReceiver(queue);
qreceiver.setMessageListener(this);
qcon.start();
}

public void close()throws JMSException
{
qreceiver.close();
qsession.close();
qcon.close();
}

public static void main(String[] args) throws Exception
{
if (args.length != 1)
{
System.out.println("Usage: java QueueReceive WebLogicURL");
return;
}
InitialContext ic = getInitialContext(args[0]);
QueueReceive qr = new QueueReceive();
qr.init(ic, QUEUE);
System.out.println("JMS Ready To Receive Messages (To quit, send a "quit" message from QueueSender.class).");
// Wait until a "quit" message has been received.
synchronized(qr)
{
while (! qr.quit)
{
try
{
qr.wait();
}
catch (InterruptedException ie)
{}
}
}
qr.close();
}

private static InitialContext getInitialContext(String url) throws NamingException
{
Hashtable env = new Hashtable();
env.put(Context.INITIAL_CONTEXT_FACTORY, JNDI_FACTORY);
env.put(Context.PROVIDER_URL, url);
return new InitialContext(env);
}
}
  4. Now open 3 – command promt Windows and run “setWLSEn.cmd” in all the three Command Windows to set the Environment (PATH & CLASSPATH).
  5. Then compile the QueueSend.java and QueueReceive.java programs.
  6. Now run the code with the following command

    7. Prompt-1

    java QueueSend t3://localhost:7003

    Prompt-2

    java QueueReceive t3://localhost:7003

    Prompt-3

    java QueueReceive t3://localhost:7004

    Note:
    With the “java QueueSend” and “java QueueReceive” command you have to give the<Protocol>://<IP-address>:<Port number>of the managed servers in different prompts.

  7. Once this is been done try to send 4 messages in the Prompt one  which is sending to the port 7003, however you will notice that the message would be received on Prompt-3 also which is listening on port 7004 as shown in the below snap-shot.

    8. .
    Thus this way we can see that even when we send messages to a particular server through UDQ the messages gets distributed evenly on all the servers in the cluster.

    .

    Thanks,

    Ravish Mody

Buffer
If you enjoyed this post, please considerleaving a comment or subscribing to the RSS feed to have future articles delivered to your feed reader.