`
fengzl
  • 浏览: 217425 次
  • 性别: Icon_minigender_1
  • 来自: 宁波
社区版块
存档分类
最新评论

JMS初探

    博客分类：
  • JMS
JMSActiveMQSecurityServletApache 
阅读更多
lib:
jms1.1.jar
activemq-all-5.0.jar

首先启动 activemq.bat或者执行以下代码启动一个broker 
import org.apache.activemq.broker.BrokerService;

/**
 * This example demonstrates how to run an embedded broker inside your Java code
 * 
 * @version $Revision: 565003 $
 */
public final class EmbeddedBroker {

    private EmbeddedBroker() {
    }

    public static void main(String[] args) throws Exception {
        BrokerService broker = new BrokerService();
        broker.setUseJmx(true);
        broker.addConnector("tcp://localhost:61616");
        broker.start();

        // now lets wait forever to avoid the JVM terminating immediately
        Object lock = new Object();
        synchronized (lock) {
            lock.wait();
        }
    }
}


消费端 
package com.jms;

import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

public class ConsumerTool implements MessageListener {
	
	 private String user = ActiveMQConnection.DEFAULT_USER;

	 private String password = ActiveMQConnection.DEFAULT_PASSWORD;

	 private String url = ActiveMQConnection.DEFAULT_BROKER_URL;

	 private String subject = "TOOL.DEFAULT";

	 private Destination destination = null;

	 private Connection connection = null;

	 private Session session = null;

	 private MessageConsumer consumer = null;

	 //初始化
	 private void initialize() throws JMSException, Exception{
		 ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user, password, url);
		 connection = connectionFactory.createConnection();
		 session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
		 destination = session.createQueue(subject);
		 consumer = session.createConsumer(destination);
	 }

	 //消费消息
	 public void consumeMessage() throws JMSException, Exception{
		 initialize();
		 connection.start();

		 System.out.println("Consumer:->Begin listening...");
		 //开始监听
		 consumer.setMessageListener(this);
		 //Message message = consumer.receive();
	 }

	 //关闭连接
	 public void close() throws JMSException{
		 System.out.println("Consumer:->Closing connection");
		 if (consumer != null)	 consumer.close();
		 if (session != null)	 session.close();
		 if (connection != null)	 connection.close();
	 }	

	public void onMessage(Message message) {
		try{
			if (message instanceof TextMessage){
				TextMessage txtMsg = (TextMessage) message;
				String msg = txtMsg.getText();
				System.out.println("Consumer:->Received: " + msg);
			} else{
				System.out.println("Consumer:->Received: " + message);
			}
		} catch (JMSException e){
			e.printStackTrace();
		}
	}

}


producter: 
package com.jms;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

public class ProducerTool {
	
	 private String user = ActiveMQConnection.DEFAULT_USER;

	 private String password = ActiveMQConnection.DEFAULT_PASSWORD;

	 private String url = ActiveMQConnection.DEFAULT_BROKER_URL;

	 private String subject = "TOOL.DEFAULT";

	 private Destination destination = null;

	 private Connection connection = null;

	 private Session session = null;

	 private MessageProducer producer = null;	

	 //初始化
	 private void initialize() throws JMSException, Exception{
		 ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user, password, url);
		 connection = connectionFactory.createConnection();
		 session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
		 destination = session.createQueue(subject);
		 producer = session.createProducer(destination);
		 producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
	 }
	 
	 //发送消息
	 public void produceMessage(String message) throws JMSException, Exception{
		 initialize();
		 TextMessage msg = session.createTextMessage(message);
		 connection.start();
		 System.out.println("Producer:->Sending message: " + message);
		 producer.send(msg);
		 System.out.println("Producer:->Message sent complete!");
	 }
	 
	 //关闭连接
	 public void close() throws JMSException{
		 System.out.println("Producer:->Closing connection");
		 if (producer != null) producer.close();
		 if (session != null) session.close();
		 if (connection != null) connection.close();
	 }	 

}


下面是一个Broker实现，不属于同一个例子！ 
package com.jms;

import java.io.IOException;
import java.util.Hashtable;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;
import javax.servlet.ServletConfig;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

public class MessageBroker extends HttpServlet
{
	private ConnectionFactory confactory=null;
	private Connection jmsCon=null;
    private Destination dest=null;
	protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException 
	{
		this.doPost(req, resp);
	}

	protected void doPost(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException 
	{
		Session session;
		try 
		{
			String deptId=null;
			String deptName=null;
			deptId=req.getParameter("deptId");
			deptName=req.getParameter("deptName");
			session = this.jmsCon.createSession(false, Session.CLIENT_ACKNOWLEDGE);
			MessageProducer msgProducer=session.createProducer(dest);
			Message textMsg=session.createTextMessage();
			textMsg.setStringProperty("deptId", deptId);
			textMsg.setStringProperty("deptName", deptName);
			msgProducer.send(textMsg);
			session.close();
			resp.sendRedirect("http://localhost/JmsTestWeb2/");
		}
		catch (Exception e)
        {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
	}
    /**
     * 
     */
	public void init(ServletConfig config) throws ServletException 
	{
		super.init(config);
		try 
		{
			this.confactory=this.getConnectionFactoryFromLdap();
			this.dest=this.getDestinationFromLdap();
			this.jmsCon=this.confactory.createConnection();
			/** 开启一个会话来创建一个消息消费者，异步监听到来的消息。*/
			Session session=jmsCon.createSession(false, Session.CLIENT_ACKNOWLEDGE);
			MessageConsumer msgConsumer=session.createConsumer(this.dest);
			MessageListenerForOrgMsg msgListener=new MessageListenerForOrgMsg();
			msgConsumer.setMessageListener(msgListener);
			/** 开启另一个会话来创建另外一个消息消费者，异步监听到来的消息。*/			 
			Session session2=jmsCon.createSession(false, Session.CLIENT_ACKNOWLEDGE);
			MessageConsumer msgConsumer2=session2.createConsumer(this.dest);
			MessageListenerForOrgMsg2 msgListener2=new MessageListenerForOrgMsg2();
			msgConsumer2.setMessageListener(msgListener2);
			this.jmsCon.start();	         
		} 
		catch (Exception e) 
		{
			e.printStackTrace();
		}
	}
	/**
	 * 
	 * @return
	 * @throws NamingException
	 */
	private ConnectionFactory getConnectionFactoryFromLdap() throws NamingException
	{
		   String account="uid=admin,ou=administrators,ou=topologymanagement,o=netscaperoot";//操作LDAP的帐户。默认就是Admin。
	       String password="111111" ;//帐户Admin的密码。
	       String root="ou=jmsstore,dc=xindongfang,dc=com"; //所操作的WLS域。也就是LDAP的根节点的DC
	       Hashtable env = new Hashtable();
	       env.put(Context.INITIAL_CONTEXT_FACTORY, "com.sun.jndi.ldap.LdapCtxFactory");//必须这样写，无论用什么LDAP服务器。
	       env.put(Context.PROVIDER_URL, "ldap://192.168.0.15:2922/" + root);//LDAP服务器的地址:端口。对WLS端口就是7001
	       env.put(Context.SECURITY_AUTHENTICATION, "simple");//授权界别，可以有三种授权级别，但是如果设为另外两种都无法登录，我也不知道为啥，但是只能设成这个值"none"。
	       env.put(Context.SECURITY_PRINCIPAL, account );//载入登陆帐户和登录密码
	       env.put(Context.SECURITY_CREDENTIALS, password);
	       InitialContext ctx=null;
	       ConnectionFactory conFacotry=null;
	       try
	       {
	    	   ctx = new InitialContext(env);//初始化上下文
	    	   conFacotry=(ConnectionFactory)ctx.lookup("cn=topicconfac");
	    	   System.out.println("get Connection factory success");
	    	   return conFacotry;
	       }
	      
	       finally
	       {
	    	   if (ctx!=null) ctx.close();   
	       }
	  }
	 /**
	  * 
	  * @return
	  * @throws NamingException
	  */
	  private Destination getDestinationFromLdap() throws NamingException
	  {
		   String account="uid=admin,ou=administrators,ou=topologymanagement,o=netscaperoot";//操作LDAP的帐户。默认就是Admin。
	       String password="111111" ;//帐户Admin的密码。
	       String root="ou=jmsstore,dc=xindongfang,dc=com"; //所操作的WLS域。也就是LDAP的根节点的DC
	       Hashtable env = new Hashtable();
	       env.put(Context.INITIAL_CONTEXT_FACTORY, "com.sun.jndi.ldap.LdapCtxFactory");//必须这样写，无论用什么LDAP服务器。
	       env.put(Context.PROVIDER_URL, "ldap://192.168.0.15:2922/" + root);//LDAP服务器的地址:端口。对WLS端口就是7001
	       env.put(Context.SECURITY_AUTHENTICATION, "simple");//授权类别，可以有三种授权级别，但是如果设为另外两种都无法登录，我也不知道为啥，但是只能设成这个值"none"。
	       env.put(Context.SECURITY_PRINCIPAL, account );//载入登陆帐户和登录密码
	       env.put(Context.SECURITY_CREDENTIALS, password);
	       InitialContext ctx=null;
	       Destination dst=null;
	       try
	       {
	    	   ctx = new InitialContext(env);//初始化上下文
	    	   dst=(Destination)ctx.lookup("cn=orgmsg");
	    	   System.out.println("get destination success");
	    	   return dst;   
	       }
	       finally
	       {
	    	   if (ctx!=null) ctx.close();
	       }
	   }

	public void destroy() 
	{
		if (this.jmsCon!=null)
		{
			try 
			{
			   this.jmsCon.close();
			} 
			catch (JMSException e) 
			{
   			   e.printStackTrace();
			}
		}
		super.destroy();
	}
}
分享到：
| JAVA多线程
评论
2 楼 fzfx88 2008-12-25  
请问 MessageListenerForOrgMsg 这个类是哪里来的?
1 楼 cddcdd 2008-04-11  

大哥你觉得ActiveMQ稳定吗？
我一直觉得不稳定，
发表评论

您还没有登录,请您登录后再发表评论

相关推荐

    java学习之spring2.0

    - **JMS和JavaMail支持**：Spring提供了对Java消息服务（JMS）和邮件服务的接口，方便开发者处理异步通信和邮件发送。 综上，Spring 2.0不仅是一个强大的框架，也是构建现代Java应用的重要基石。它的全面性和灵活...

    spring-framework-0.9.1.zip

    《Spring Framework 0.9.1：初探与解析》 Spring Framework，作为Java开发领域中的一个里程碑式框架，自其诞生以来就以其强大的企业级应用支持和灵活的编程模式深受开发者喜爱。本文将深入探讨Spring Framework ...

    Spring培训资料

    **2.5 初探PAFA** 通过简单的“Hello World”示例，让开发者快速了解PAFA的基本用法和流程。 #### 三、PAFA应用的整体结构 **3.1 开发目录布局** PAFA应用通常包含以下目录结构： - `src/main/java`：存放Java源...

    Spring3.x权威开发指南

    Spring支持与Java Message Service (JMS) 的集成，使得开发者可以使用消息传递中间件来实现异步通信。 **7.5 集成JavaMail** Spring支持与JavaMail的集成，使得开发者可以轻松地发送电子邮件。 综上所述，...

    Project-Mule1

    【标题】：Mule ESB 项目初探 在IT领域，企业服务总线（Enterprise Service Bus，ESB）是一种架构模式，用于实现不同应用程序之间的松耦合通信。Mule ESB 是一个开源的ESB产品，由MuleSoft公司开发，它提供了强大的...

Magicbox
Global site tag (gtag.js) - Google Analytics