程序员经验分享-hwhale

JMS消息监听器Weblogic的并发处理

2024-04-17

我正在 JMS 上运行测试用例，发现处理是顺序的。当我向使用 JMS 发送消息的 servlet 发出 200 个请求时，接收器（messageListner）正在按顺序接收请求。如何接收并发请求？我们有什么参数要设置吗？我阅读了 JMS 教程和 API，在同一会话中消息是按顺序传递的，即使我为每个发送请求创建一个新会话，并且在接收端的 10 个会话仍然处理是连续的。

public class ProducerServlet extends javax.servlet.http.HttpServlet implements
    javax.servlet.Servlet {

// Defines the JNDI context factory.
public final static String JNDI_FACTORY = "weblogic.jndi.WLInitialContextFactory";

// Defines the JMS context factory.
public final static String JMS_FACTORY = "jms/TestConnectionFactory";

// Defines the queue.
public final static String QUEUE = "jms/TestJMSQueue";

public final static String TOPIC = "jms/TestTopic";

TestJMSListener jms = new TestJMSListener();
ConnectionFactory connectionFactory = null;
Queue dest1 = null;
Topic dest =null;
Connection connection = null;
MessageProducer producer = null;

protected void doGet(HttpServletRequest request,
        HttpServletResponse response) throws ServletException, IOException {
        try {
            connection = connectionFactory.createConnection();              

        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        producer = session.createProducer(dest1);
        TextMessage message = session.createTextMessage();

        message.setText("This is message from JMSSECOND DEMO "
                + request.getParameter("Num"));
        System.out.println("Sending message: " + message.getText());
        producer.send(message);
        producer.send(session.createMessage());
    } catch (Exception e) {
        System.out.println("Exception occurred: " + e.toString());
    }

}

@Override
public void init(ServletConfig arg0) throws ServletException {      
    Context jndiContext = null;
    try {

        Hashtable env = new Hashtable();
        env.put(Context.INITIAL_CONTEXT_FACTORY, JNDI_FACTORY);
        env.put(Context.PROVIDER_URL, "http://localhost:7001");
        jndiContext = new InitialContext(env);
    } catch (NamingException e) {
        System.out.println("Could not create JNDI API context: "
                + e.toString());            
    }

    try {
        connectionFactory = (ConnectionFactory) jndiContext
                .lookup(JMS_FACTORY);
        dest1 = (Queue) jndiContext.lookup(QUEUE);
    } catch (Exception e) {
        System.out.println("JNDI API lookup failed: " + e.toString());
        e.printStackTrace();            
    }

}

}

监听器实现，在收到消息后我要睡觉（做一秒钟的事情）。

public class TestJMSListener implements MessageListener {

// Defines the JNDI context factory.
public final static String JNDI_FACTORY = "weblogic.jndi.WLInitialContextFactory";

// Defines the JMS context factory.
public final static String JMS_FACTORY = "jms/TestConnectionFactory";

// Defines the queue.
public final static String QUEUE = "jms/TestJMSQueue";

public final static String TOPIC = "jms/TestTopic";

public TestJMSListener() {

    System.out.println("********* Consumer check **********");

    Context jndiContext = null;
    ConnectionFactory connectionFactory = null;
    Connection connection[] = null;
    Session session[] = null;
    Queue dest1 = null;
    Topic dest = null;
    MessageConsumer consumer[] = null;
    // TextMessage message = null;

    try {
        Hashtable env = new Hashtable();
        env.put(Context.INITIAL_CONTEXT_FACTORY, JNDI_FACTORY);
        env.put(Context.PROVIDER_URL, "http://localhost:7001");
        jndiContext = new InitialContext(env);
    } catch (NamingException e) {
        System.out.println("Could not create JNDI API context: "
                + e.toString());
        System.exit(1);
    }

    try {
        connectionFactory = (ConnectionFactory) jndiContext
                .lookup(JMS_FACTORY);
        dest1 = (Queue) jndiContext.lookup(QUEUE);
    } catch (Exception e) {
        System.out.println("JNDI API lookup failed: " + e.toString());
        System.exit(1);
    }
    connection = new Connection[10];
    session = new Session[10];
    consumer = new MessageConsumer[10];
    for (int i = 0; i < 10; i++) {
        try {

            connection[i] = connectionFactory.createConnection();
            session[i] = connection[i].createSession(false,
                    Session.AUTO_ACKNOWLEDGE);
            consumer[i] = session[i].createConsumer(dest);
            consumer[i].setMessageListener(this);
            connection[i].start();
        } catch (JMSException e) {
            System.out.println("Exception occurred: " + e.toString());
        }
    }
}

@Override
public void onMessage(Message m) {

    if (m instanceof TextMessage) {
        TextMessage message = (TextMessage) m;
        try {
            System.out.println("Reading message from Listener: "
                    + new Date() + message.getText());
            Thread.sleep(1000);
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

}

 }

我使用的是 Weblogic 11g，具有 ConnectionFactory 和 Queue 的默认配置。当我使用 Topic 时，它实际上每秒仅传送一条消息（即在第一条消息完成后），而对于 Queue，它每秒传送 2 到 3 条消息。如何让我的监听器支持并发处理。

最终解决方案

添加更多侦听器对象而不是侦听器中的多个会话/消费者，它解决了目的。 找到下面的修改后的代码。

public class ProducerServlet extends javax.servlet.http.HttpServlet implements
    javax.servlet.Servlet {

// Defines the JNDI context factory.
public final static String JNDI_FACTORY = "weblogic.jndi.WLInitialContextFactory";

// Defines the JMS context factory.
public final static String JMS_FACTORY = "jms/TestConnectionFactory";

// Defines the queue.
public final static String QUEUE = "jms/TestJMSQueue";

public final static String TOPIC = "jms/TestTopic";
TestJMSListener listeners[] = new TestJMSListener[20];
ConnectionFactory connectionFactory = null;
Queue dest1 = null;
Topic dest =null;
Connection connection = null;
MessageProducer producer = null;

protected void doGet(HttpServletRequest request,
        HttpServletResponse response) throws ServletException, IOException {
        try {
            connection = connectionFactory.createConnection();              

        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        producer = session.createProducer(dest1);
        TextMessage message = session.createTextMessage();

        message.setText("This is message from JMSSECOND DEMO "
                + request.getParameter("Num"));
        System.out.println("Sending message: " + message.getText());
        producer.send(message);
        producer.send(session.createMessage());
    } catch (Exception e) {
        System.out.println("Exception occurred: " + e.toString());
    }

}

@Override
public void init(ServletConfig arg0) throws ServletException {      
    Context jndiContext = null;
    try {

        Hashtable env = new Hashtable();
        env.put(Context.INITIAL_CONTEXT_FACTORY, JNDI_FACTORY);
        env.put(Context.PROVIDER_URL, "http://localhost:7001");
        jndiContext = new InitialContext(env);
    } catch (NamingException e) {
        System.out.println("Could not create JNDI API context: "
                + e.toString());            
    }

    try {
        connectionFactory = (ConnectionFactory) jndiContext
                .lookup(JMS_FACTORY);
        dest1 = (Queue) jndiContext.lookup(QUEUE);
        for(int i=0;i<listeners.length;i++ ){
        listeners[i]=new TestJMSListener(Integer.toString(i+1));    
        }

    } catch (Exception e) {
        System.out.println("JNDI API lookup failed: " + e.toString());
        e.printStackTrace();            
    }

}

}


public class TestJMSListener implements MessageListener {

// Defines the JNDI context factory.
public final static String JNDI_FACTORY = "weblogic.jndi.WLInitialContextFactory";

// Defines the JMS context factory.
public final static String JMS_FACTORY = "jms/TestConnectionFactory";

// Defines the queue.
public final static String QUEUE = "jms/TestJMSQueue";

public final static String TOPIC = "jms/TestTopic";

public String listnerNum = "";
public TestJMSListener(String listerNo) {
    super();
    System.out.println("********* Consumer check **********");
    listnerNum = listerNo;
    Context jndiContext = null;
    ConnectionFactory connectionFactory = null;
    Connection connection = null;
    Session session = null;
    Queue dest1 = null;
    Topic dest = null;
    MessageConsumer consumer = null;
    // TextMessage message = null;

    try {
        Hashtable env = new Hashtable();
        env.put(Context.INITIAL_CONTEXT_FACTORY, JNDI_FACTORY);
        env.put(Context.PROVIDER_URL, "http://localhost:7001");
        jndiContext = new InitialContext(env);
    } catch (NamingException e) {
        System.out.println("Could not create JNDI API context: "
                + e.toString());
        System.exit(1);
    }

    try {
        connectionFactory = (ConnectionFactory) jndiContext
                .lookup(JMS_FACTORY);
        dest1 = (Queue) jndiContext.lookup(QUEUE);
    } catch (Exception e) {
        System.out.println("JNDI API lookup failed: " + e.toString());
        System.exit(1);
    }
    try{
            connection = connectionFactory.createConnection();
            session = connection.createSession(false,
                    Session.AUTO_ACKNOWLEDGE);
            consumer = session.createConsumer(dest1);
            consumer.setMessageListener(this);
            connection.start();
        } catch (JMSException e) {
            System.out.println("Exception occurred: " + e.toString());
        }


}

@Override
public void onMessage(Message m) {

    if (m instanceof TextMessage) {
        TextMessage message = (TextMessage) m;
        try {
            System.out.println("Reading message from Listener: "+listnerNum+ " : "
                    + new Date() + message.getText());
            Thread.sleep(1000);
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

}

}

在您的代码中，您只有一个 Listener 实例（在创建 Servlet 实例时创建），因此您将仅顺序接收消息，
无论您有多少发送者会话..它都是队列。
如果你想同时接收，那么你可能需要多个监听器，并且任一监听器中只会传递一次消息。
如果您想同时处理消息，一旦按顺序传递消息，则创建线程池并在单独的线程中委托该进程并返回到侦听模式。
注意** 在此模式下，您可能无法正确处理 Ack 模式，因为您在未完成消息过程的情况下就进行了确认。

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)

Java

concurrency

JMS

JMS消息监听器Weblogic的并发处理 的相关文章

随机推荐

热门标签