码迷，mamicode.com
首页 > 其他好文 > 详细

Intellij IDEA 创建消息驱动Bean - 接收JMS消息

时间：2016-04-01 20:36:34      阅读：187      评论：0      收藏：0      [点我收藏+]

标签：

除了同步方式的调用之外，有时还需要异步调用，用来处理不需要即时处理的信息，例如短信、邮件等，这需要使用EJB中的独特组件——消息驱动Bean（Message-Driven Bean，MDB），它提供了Java消息服务JMS（Java Messaging Service）的处理能力，由消息驱动Bean来处理JMS消息。JMS的消息由客户端程序产生，并被发布到服务器的消息队列，消息驱动Bean随之检索消息并执行其内容。这种事件或者数据的通信就称为异步形式，客户端或者服务端Bean都无须依赖对方的直接响应。
 

参考：

2）JBoss 7/WildFly中配置使用JMS消息队列：http://www.tuicool.com/articles/Vf6bYv
3）Eclipse+JBoss+EJB3消息驱动Bean：http://developer.51cto.com/art/200906/131361.htm
6）jboss 7.1.1 消息驱动bean配置：http://jeremy102.iteye.com/blog/1706852
7）Jboss7 JMS demo：http://lym6520.iteye.com/blog/1600630
 
构建JMB的消息驱动Bean（Message-Driven Bean）有4中方式：
  • Management Console
  • Management CLI
  • 部署*-jms.xml 文件到 deployments目录
  • 编辑JBoss配置文件
JMS消息的两种类型：队列（Queue）与主题（Topic）
在JMS中有以下两种类型的目的地（Destination）：
· 队列（Queue）：用于保存从一个JMS客户端发往另一个JMS客户端的消息，这种消息模式被称为点对点（Point-To-Point）模式；
· 主题（Topic）：用于保存从许多潜在的JMS客户端发往多个潜在的JMS客户端的消息，这种消息模型被称为发布/订阅（Public/Subscribe）模式。
 
使用最简单的创建MDB的方法：编辑JBoss配置文件的方法。
 
1、找到JBoss安装路径的这个文件：C:\jboss-as-7.1.1.Final\standalone\configuration\standalone-full.xml。找到
<hornetq-server>
　　<jms-destinations>
　　</jms-destinations>
</hornetq-server>
在中间加上新创建的MDB的内容：
<jms-queue name="HelloWorldMDBQueue">
    <entry name="java:jboss/exported/jms/queue/HelloWorldMDBQueue"/>
</jms-queue>
<jms-topic name="HelloWorldMDBTopic">
    <entry name="java:jboss/exported/jms/topic/HelloWorldMDBTopic"/>
</jms-topic>
 
2、使用命令行方式启动JBoss：
切换到JBoss安装路径下的bin目录下：cd C:\jboss-as-7.1.1.Final\bin
启动命令：standalone.bat --server-config=standalone-full.xml  
 
3、编写队列模式的测试代码：
1）消息生成者：
package com.ejb.mdb;
import javax.jms.*;
import javax.naming.Context;
import javax.naming.InitialContext;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.logging.Logger;
/**
 * Created by gao on 16-3-28.
 */
public class JMSProducer {
    private static final Logger log = Logger.getLogger(JMSProducer.class.getName());
    private static final String DEFAULT_MESSAGE = "这是JMS信息.....";
    private static final String DEFAULT_CONNECTION_FACTORY = "jms/RemoteConnectionFactory";
    private static final String DEFAULT_DESTINATION = "/jms/queue/HelloWorldMDBQueue";
    private static final String DEFAULT_MESSAGE_COUNT = "10";
    private static final String DEFAULT_USERNAME = "yuqin";
    private static final String DEFAULT_PASSWORD = "123456";
    private static final String INITIAL_CONTEXT_FACTORY = "org.jboss.naming.remote.client.InitialContextFactory";
    private static final String PROVIDER_URL = "remote://localhost:4447";
    public static void main(String[] args) throws Exception {
        Context context=null;
        Connection connection=null;
        try {
            // 设置上下文的JNDI查找
            log.info("设置JNDI访问环境信息也就是设置应用服务器的上下文信息!");
            final Properties env = new Properties();
            env.put(Context.INITIAL_CONTEXT_FACTORY, INITIAL_CONTEXT_FACTORY);// 该KEY的值为初始化Context的工厂类,JNDI驱动的类名
            env.put(Context.PROVIDER_URL, PROVIDER_URL);// 该KEY的值为Context服务提供者的URL.命名服务提供者的URL
            env.put(Context.SECURITY_PRINCIPAL, DEFAULT_USERNAME);
            env.put(Context.SECURITY_CREDENTIALS, DEFAULT_PASSWORD);//应用用户的登录名,密码.
            // 获取到InitialContext对象.
            context = new InitialContext(env);
            log.info("初始化上下文,‘JNDI驱动类名‘,‘服务提供者URL‘,‘应用用户的账户‘,‘密码‘完毕.");
            log.info("获取连接工厂!");
            ConnectionFactory connectionFactory = (ConnectionFactory) context.lookup(DEFAULT_CONNECTION_FACTORY);
            log.info("获取目的地!");
            Destination destination = (Destination) context.lookup(DEFAULT_DESTINATION);
            // 创建JMS连接、会话、生产者和消费者
            connection = connectionFactory.createConnection(DEFAULT_USERNAME, DEFAULT_PASSWORD);
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            MessageProducer producer = session.createProducer(destination);
            connection.start();
            int count = Integer.parseInt(DEFAULT_MESSAGE_COUNT);
            // 发送特定数目的消息
            TextMessage message = null;
            for (int i = 0; i < count; i++) {
                message = session.createTextMessage(DEFAULT_MESSAGE);
                producer.send(message);
                log.info("message:"+message);
                log.info("message:"+DEFAULT_MESSAGE);
            }
            // 等待30秒退出
            CountDownLatch latch = new CountDownLatch(1);
            latch.await(30, TimeUnit.SECONDS);
        } catch (Exception e) {
            log.severe(e.getMessage());
            throw e;
        } finally {
            if (context != null) {
                context.close();
            }
            // 关闭连接负责会话,生产商和消费者
            if (connection != null) {
                connection.close();
            }
        }
    }
}

 2）消息消费者：

package com.ejb.mdb;
import javax.jms.*;
import javax.naming.Context;
import javax.naming.InitialContext;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.logging.Logger;
/**
 * Created by gao on 16-3-28.
 */
public class JMSConsumer {
    private static final Logger log = Logger.getLogger(JMSConsumer.class.getName());
    private static final String DEFAULT_CONNECTION_FACTORY = "jms/RemoteConnectionFactory";
    private static final String DEFAULT_DESTINATION = "/jms/queue/HelloWorldMDBQueue";
    private static final String DEFAULT_USERNAME = "yuqin";
    private static final String DEFAULT_PASSWORD = "123456";
    private static final String INITIAL_CONTEXT_FACTORY = "org.jboss.naming.remote.client.InitialContextFactory";
    private static final String PROVIDER_URL = "remote://localhost:4447";
    private static final int WAIT_COUNT = 5;
    public static void main(String[] args) throws Exception {
        ConnectionFactory connectionFactory = null;
        Connection connection = null;
        Session session = null;
        MessageConsumer consumer = null;
        Destination destination = null;
        TextMessage message = null;
        Context context = null;
        try {
            final Properties env = new Properties();
            env.put(Context.INITIAL_CONTEXT_FACTORY, INITIAL_CONTEXT_FACTORY);
            env.put(Context.PROVIDER_URL, PROVIDER_URL);
            env.put(Context.SECURITY_PRINCIPAL, DEFAULT_USERNAME);
            env.put(Context.SECURITY_CREDENTIALS, DEFAULT_PASSWORD);
            context = new InitialContext(env);
            connectionFactory = (ConnectionFactory) context.lookup(DEFAULT_CONNECTION_FACTORY);
            destination = (Destination) context.lookup(DEFAULT_DESTINATION);
            connection = connectionFactory.createConnection(DEFAULT_USERNAME, DEFAULT_PASSWORD);
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            consumer = session.createConsumer(destination);
            connection.start();
            // 等待30秒退出
            CountDownLatch latch = new CountDownLatch(1);
            log.info("开始从JBOSS端接收信息-----");
            int i = 0;
            for (; i < WAIT_COUNT; i++) {
                if (message != null) {
                    log.info("接收到的消息的内容:" + message.getText());
                    i = 0;
                }
                log.info("开始从JBOSS端接收信息-----");
                message = (TextMessage) consumer.receive(5000);
                latch.await(1, TimeUnit.SECONDS);
            }
        } catch (Exception e) {
            log.severe(e.getMessage());
            throw e;
        } finally {
            if (context != null) {
                context.close();
            }
            if (connection != null) {
                connection.close();
            }
        }
    }
}
3）运行：
先运行消息消费者，用来监听消息，然后再运行消息生成者。
 
 
4、编写主题模式的测试代码：
1）消息订阅者：
package com.ejb.mdb;
import javax.jms.*;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.logging.Logger;
/**
 * Created by gao on 16-3-28.
 */
public class JMSSub {
    private static final String DEFAULT_USERNAME = "yuqin";
    private static final String DEFAULT_PASSWORD = "123456";
    private static final String INITIAL_CONTEXT_FACTORY = "org.jboss.naming.remote.client.InitialContextFactory";
    private static final String PROVIDER_URL = "remote://localhost:4447";
    private static final String DEFAULT_CONNECTION_FACTORY = "jms/RemoteConnectionFactory";
    private static final Logger log = Logger.getLogger(JMSSub.class.getName());
    /**
     * @param args
     * @throws Exception
     */
    public static void main(String[] args) throws Exception {
        ConnectionFactory connectionFactory = null;
        Connection connection = null;
        Session session = null;
        Topic topic = null;
        Context context = null;
        MessageConsumer consumer = null;
        try {
            log.info("设置JNDI访问环境信息也就是设置应用服务器的上下文信息!");
            final Properties env = new Properties();
            env.put(Context.INITIAL_CONTEXT_FACTORY, INITIAL_CONTEXT_FACTORY);
            env.put(Context.PROVIDER_URL, PROVIDER_URL);
            env.put(Context.SECURITY_PRINCIPAL, DEFAULT_USERNAME);
            env.put(Context.SECURITY_CREDENTIALS, DEFAULT_PASSWORD);
            context = new InitialContext(env);
            log.info("初始化上下文,‘JNDI驱动类名‘,‘服务提供者URL‘,‘应用用户的账户‘,‘密码‘完毕.");
            log.info("创建JMS连接、会话、主题!");
            connectionFactory = (ConnectionFactory) context.lookup(DEFAULT_CONNECTION_FACTORY);
            connection = connectionFactory.createConnection(DEFAULT_USERNAME, DEFAULT_PASSWORD);
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            topic = session.createTopic("HelloWorldMDBTopic");
            consumer = session.createConsumer(topic);
            consumer.setMessageListener(new javax.jms.MessageListener() {
                public void onMessage(Message message) {
                    try {
                        TextMessage tm = (TextMessage) message;
                        System.out.println("接收到的消息内容: " + tm.getText().toString());
                        System.out.println("JMS目的地: " + tm.getJMSDestination());
                        System.out.println("JMS回复: " + tm.getJMSReplyTo());
                        System.out.println("JMS消息ID号: " + tm.getJMSMessageID());
                        System.out.println("是否重新接收: " + tm.getJMSRedelivered());
                    } catch (JMSException e1) {
                        e1.printStackTrace();
                    }
                }
            });
            connection.start();
            //等待30秒退出
            CountDownLatch latch = new CountDownLatch(1);
            latch.await(100, TimeUnit.SECONDS);
        } catch (Exception e) {
            log.severe(e.getMessage());
            throw e;
        } finally {
            if (context != null) {
                context.close();
            }
            // 关闭连接负责会话,发布者和订阅者
            if (connection != null) {
                connection.close();
            }
        }
    }
}

 2）消息发布者：

package com.ejb.mdb;
import javax.jms.*;
import javax.naming.Context;
import javax.naming.InitialContext;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.Properties;
import java.util.logging.Logger;
/**
 * Created by gao on 16-3-28.
 */
public class JMSPub {
    private static final String DEFAULT_USERNAME = "yuqin";
    private static final String DEFAULT_PASSWORD = "123456";
    private static final String INITIAL_CONTEXT_FACTORY = "org.jboss.naming.remote.client.InitialContextFactory";
    private static final String PROVIDER_URL = "remote://localhost:4447";
    private static final String DEFAULT_CONNECTION_FACTORY = "jms/RemoteConnectionFactory";
    private static final String DEFAULT_DESTINATION = "/jms/topic/HelloWorldMDBTopic";
    private static final Logger log = Logger.getLogger(JMSPub.class.getName());
    public static void main(String[] args) throws Exception {
        ConnectionFactory connectionFactory = null;
        Connection connection = null;
        Session session = null;
        Topic topic = null;
        Context context = null;
        MessageProducer producer = null;
        BufferedReader msgStream = null;
        try {
            log.info("设置JNDI访问环境信息也就是设置应用服务器的上下文信息!");
            final Properties env = new Properties();
            env.put(Context.INITIAL_CONTEXT_FACTORY, INITIAL_CONTEXT_FACTORY);
            env.put(Context.PROVIDER_URL, PROVIDER_URL);
            env.put(Context.SECURITY_PRINCIPAL, DEFAULT_USERNAME);
            env.put(Context.SECURITY_CREDENTIALS, DEFAULT_PASSWORD);
            context = new InitialContext(env);
            log.info("初始化上下文,‘JNDI驱动类名‘,‘服务提供者URL‘,‘应用用户的账户‘,‘密码‘完毕.");
            log.info("获取连接工厂!");
            connectionFactory = (ConnectionFactory) context.lookup(DEFAULT_CONNECTION_FACTORY);
            log.info("创建JMS连接、会话、主题!");
            connection = connectionFactory.createConnection(DEFAULT_USERNAME, DEFAULT_PASSWORD);
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            topic = (Topic) context.lookup(DEFAULT_DESTINATION);
            producer = session.createProducer(topic);
            msgStream = new BufferedReader(new InputStreamReader(System.in));
            String line = null;
            boolean quitNow = false;
            do {
                System.out.print("输入要发送的消息：(数字0退出)");
                line = msgStream.readLine();
                if (line != null && line.trim().length() != 0) {
                    TextMessage textMessage = session.createTextMessage();
                    textMessage.setText(line);
                    producer.send(textMessage);
                    quitNow = line.equalsIgnoreCase("0");
                }
            } while (!quitNow);
        } catch (Exception e) {
            log.severe(e.getMessage());
            throw e;
        } finally {
            if (context != null) {
                context.close();
            }
            // 关闭连接负责会话,发布者和订阅者
            if (connection != null) {
                connection.close();
            }
        }
    }
}
3）运行：
先运行消息订阅者，监听消息发布者的传过来的消息；然后再运行消息发布者。
 
 
5、项目结构图：
技术分享
 
 
 
 
 
 
 
 
 

Intellij IDEA 创建消息驱动Bean - 接收JMS消息

标签：

原文地址：http://www.cnblogs.com/yangyquin/p/5346104.html
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
登录后才能评论！
分享档案
更多>
2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)
周排行
mamicode.com排行更多图片
更多
友情链接
兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
关于我们 - 联系我们 - 留言反馈
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码！