/** * <html> * <body> * <P> https://github.com/Jasonandy </p> * <p> All rights reserved.</p> * <p> Created on 2018年6月12日 下午4:00:58</p> * <p> Created by Jason</p> * </body> * </html> */ package cn.ucaner.component.jms; import java.util.Hashtable; import javax.jms.JMSException; import javax.jms.Session; import javax.jms.TextMessage; import javax.jms.Topic; import javax.jms.TopicConnection; import javax.jms.TopicConnectionFactory; import javax.jms.TopicPublisher; import javax.jms.TopicSession; import javax.naming.Context; import javax.naming.InitialContext; import javax.naming.NamingException; /** * @Package：cn.ucaner.component.jms * @ClassName：MsgPublisher * @Description： <p> jms publiser </p> * <url> https://blog.csdn.net/zhangzikui/article/details/24837999 </url> * @Author： - Jason * @CreatTime：2018年6月12日 下午4:00:58 * @Modify By： * @ModifyTime： 2018年6月12日 * @Modify marker： * @version V1.0 */ public class MsgPublisher { /** * JMS连接，属于Pub/Sub方式的连接 */ TopicConnection topicConnection; /** * JMS会话，属于Pub/Sub方式的会话 */ TopicSession topicSession; /** * 消息发布者 */ TopicPublisher topicPublisher; /** * 主题 */ Topic topic; public MsgPublisher(String factoryJNDI, String topicJNDI)throws JMSException, NamingException { Hashtable<String, String> env = new Hashtable<String, String>(); /** * 设置好连接JMS容器的属性，不同的容器需要的属性可能不同，需要查阅相关文档 */ env.put(Context.INITIAL_CONTEXT_FACTORY, "org.jnp.interfaces.NamingContextFactory"); env.put(Context.PROVIDER_URL, "localhost:1088"); env.put("java.naming.rmi.security.manager", "yes"); env.put(Context.URL_PKG_PREFIXES, "org.jboss.naming"); /** * 创建连接JMS容器的上下文(context) */ Context context = new InitialContext(env); /** * 通过连接工厂的JNDI名查找ConnectionFactory */ TopicConnectionFactory topicFactory = (TopicConnectionFactory) context.lookup(factoryJNDI); /** * 用连接工厂创建一个JMS连接 */ topicConnection = topicFactory.createTopicConnection(); /** * 通过JMS连接创建一个Session */ topicSession = topicConnection.createTopicSession(false,Session.AUTO_ACKNOWLEDGE); /** * 通过上下文查找到一个主题(topic) */ topic = (Topic) context.lookup(topicJNDI); /** * 用session来创建一个特定主题的消息发送者 */ topicPublisher = topicSession.createPublisher(topic); } /** * @Description: 发布一条文本消息 * @param msg 待发布的消息 * @throws JMSException void * @Autor: Jason - jasonandy@hotmail.com */ public void publish(String msg) throws JMSException { /** * 用session来创建一个文本类型的消息 */ TextMessage message = topicSession.createTextMessage(); /** * 设置消息内容 */ message.setText(msg); /** * 消息发送，发送到特定主题 */ topicPublisher.publish(topic, message); } public void close() throws JMSException { topicSession.close();//关闭session topicConnection.close();//关闭连接 } public static void main(String[] args)throws JMSException, NamingException { MsgPublisher publisher = new MsgPublisher("ConnectionFactory", "topic/testTopic"); try { for (int i = 1; i < 11; i++) { String msg = "Hello World no. " + i; System.out.println("Publishing message: " + msg); publisher.publish(msg); } /** * session和connection用完之后一定记得关闭 */ publisher.close(); } catch (Exception ex) { ex.printStackTrace(); } } }
/** * <html> * <body> * <P> https://github.com/Jasonandy </p> * <p> All rights reserved.</p> * <p> Created on 2018年6月12日 下午4:00:58</p> * <p> Created by Jason</p> * </body> * </html> */ package cn.ucaner.component.jms; import java.util.Hashtable; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageListener; import javax.jms.Session; import javax.jms.TextMessage; import javax.jms.Topic; import javax.jms.TopicConnection; import javax.jms.TopicConnectionFactory; import javax.jms.TopicSession; import javax.jms.TopicSubscriber; import javax.naming.Context; import javax.naming.InitialContext; import javax.naming.NamingException; /** * @Package：cn.ucaner.component.jms * @ClassName：MsgSubscriber * @Description： <p> MsgSubscriber </p> * <url> https://blog.csdn.net/zhangzikui/article/details/24837999 </url> * @Author： - * @CreatTime：2018年6月12日 下午4:10:55 * @Modify By： * @ModifyTime： 2018年6月12日 * @Modify marker： * @version V1.0 */ public class MsgSubscriber implements MessageListener{ TopicConnection topicConnection; TopicSession topicSession; TopicSubscriber topicSubscriber; Topic topic; @Override public void onMessage(Message msg) { try { if (msg instanceof TextMessage) { //把Message 转型成 TextMessage 并提取消息内容 String msgTxt = ((TextMessage) msg).getText(); System.out.println("HelloSubscriber got message: " +msgTxt); } } catch (JMSException ex) { System.err.println("Could not get text message: " + ex); ex.printStackTrace(); } } public MsgSubscriber(String factoryJNDI, String topicJNDI)throws JMSException, NamingException { Hashtable<String, String> env = new Hashtable<String, String>(); //设置好连接JMS容器的属性，不同的容器需要的属性可能不同，需要查阅相关文档 env.put(Context.INITIAL_CONTEXT_FACTORY,"org.jnp.interfaces.NamingContextFactory"); env.put(Context.PROVIDER_URL, "localhost:1088"); env.put("java.naming.rmi.security.manager", "yes"); env.put(Context.URL_PKG_PREFIXES, "org.jboss.naming"); Context context = new InitialContext(); TopicConnectionFactory topicFactory = (TopicConnectionFactory) context.lookup(factoryJNDI); //创建连接 topicConnection = topicFactory.createTopicConnection(); topicSession = topicConnection.createTopicSession(false,Session.AUTO_ACKNOWLEDGE);//创建session topic = (Topic) context.lookup(topicJNDI);//查找到主题 //用session创建一个特定queue的消息接收者 topicSubscriber = topicSession.createSubscriber(topic); //注册监听，这里设置的监听是自己，因为本类已经实现了MessageListener接口， //一旦queueReceiver接收到了消息，就会调用本类的onMessage方法 topicSubscriber.setMessageListener(this); System.out.println("HelloSubscriber subscribed to topic: "+ topicJNDI); topicConnection.start();//启动连接，这时监听器才真正生效 } public void close() throws JMSException { topicSession.close(); topicConnection.close(); } public static void main(String[] args) { try { new MsgSubscriber("TopicConnectionFactory","topic/testTopic"); } catch (Exception ex) { ex.printStackTrace(); } } }