Intellij IDEA 创建消息驱动Bean - 接收JMS消息
Posted 我是一名老菜鸟
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Intellij IDEA 创建消息驱动Bean - 接收JMS消息相关的知识，希望对你有一定的参考价值。
除了同步方式的调用之外，有时还需要异步调用，用来处理不需要即时处理的信息，例如短信、邮件等，这需要使用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消息的主要内容，如果未能解决你的问题，请参考以下文章
IntelliJ IDEA Maven 整合SSH遇到的问题，及application的配置
IntelliJ IDEA Spring+Mybatis dao bean对象注入失败
Intellij Idea - 无法自动装配。找不到“ApplicationRepository”类型的 bean。 less... (Ctrl+F1) 检查 bean 类中的自动装配问题
Builgen 插件——IntelliJ IDEA和Eclipse Java Bean Builder模式代码生成器-比lombok更符合需求