当前位置:首页 > 开发 > 编程语言 > 消息中间件 > 正文

JMS Api Demo

发表于: 2012-12-28   作者:antlove   来源:转载   浏览次数:
摘要: jms.JMSFactory package jms; import javax.jms.TopicConnectionFactory; import org.apache.activemq.ActiveMQConnectionFactory; public class JMSFactory { private static ActiveMQConnectionFactory
jms.JMSFactory
package jms;
import javax.jms.TopicConnectionFactory;
import org.apache.activemq.ActiveMQConnectionFactory;
public class JMSFactory {
	private static ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_BROKER_BIND_URL);
	
	public static TopicConnectionFactory getActiveMQConnectionFactory(){
		return activeMQConnectionFactory;
	}
}

jms.JMSMessageActor
package jms;
import javax.jms.*;
public abstract class JMSMessageActor {
	protected String name=null;
	protected String defaultQueueName="defalut-queue";
	protected String defaultTopicName="defalut-topic";
	protected final int  DESTIONATION_TYPE_TOPIC=1;
	protected final int DESTIONATION_TYPE_QUEUE=2;
	public JMSMessageActor(String name) {
		this.name = name;
	}
	public abstract Destination  getDestination();
	public Destination createDefaultDestination(int type){
		Destination dest=null;
		switch(type){
			case DESTIONATION_TYPE_TOPIC:
				dest=new Topic(){
					@Override
					public String getTopicName() throws JMSException {
						return defaultTopicName;
					}};break;
			case DESTIONATION_TYPE_QUEUE:
				dest=new Queue(){
					@Override
					public String getQueueName() throws JMSException {
						return defaultQueueName;
					}};
				break;
		}
		return dest;
	}
}

jms.JMSMessageConsumer
package jms;
import javax.jms.*;
public abstract class JMSMessageConsumer extends JMSMessageActor implements Runnable,MessageListener{
	public JMSMessageConsumer(String name){
		super(name);
	}
	@Override
	public void onMessage(Message message) {
		synchronized(JMSMessageConsumer.class){
			System.out.println("##### consumer "+ name +" receive message. #####");
			System.out.println(JMSUtil.formatMessage(message));
		}
	}
	@Override
	public void run() {
		try{			
			// get topic connect factory
			ConnectionFactory factory = JMSFactory.getActiveMQConnectionFactory();
			// create connection
			Connection connection = factory.createConnection();
			// create unique client id for the connection
			connection.setClientID("consumer_connection_"+name);
			// if the connection start method is not invoked , the consumer may be not receive the message
			connection.start();
			// create session
			Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
			Destination destination=getDestination();
			// if the destination is an instance of Queue , 
			// it will receive the message from the queue,
			// in other words the message can be consumed one time by one consumer 
			// and the message is durable.
			// if the destination is an instance of Topic , 
			// the subscribers of the Topic can receive the message,
			// but the message is non durable.
			MessageConsumer consumer  =session.createConsumer(destination,null,true);
			// if the destination is an instance of Topic, 
			// specify the clientID of the connection
			// and create MessageConsumer like this,
			// the subscribers of the Topic can receive the message
			// and the message is durable.
			//consumer =session.createDurableSubscriber((Topic)destination, "durable topic", null, true);
			consumer.setMessageListener(this);
		}catch(Exception e){
			throw new RuntimeException(e);
		}
	}
}

jms.JMSMessageProducer
package jms;
import javax.jms.*;
public abstract class JMSMessageProducer extends JMSMessageActor implements Runnable{
	public JMSMessageProducer(String name){
		super(name);
	}
	@Override
	public void run() {
		try{
			// get topic connect factory
			ConnectionFactory factory = JMSFactory.getActiveMQConnectionFactory(); 
			// create connection
			Connection connection = factory.createConnection();
			// create session
			Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
			Destination destination=getDestination();
			// create message producer
			MessageProducer producer = session.createProducer(destination);
			// create message
			MapMessage mapMessage=session.createMapMessage();
			mapMessage.setObjectProperty("lock", "key");
			mapMessage.setObjectProperty("vegetables", "greens");
			mapMessage.setObjectProperty("fruit", "apple");
			mapMessage.setObjectProperty("meat", "pork");
			
			// producer.setTimeToLive(1000);// set the message expiration time
			
			producer.send(mapMessage);// send message
			connection.close();
			System.out.println(name+" send message success!");
		}catch(Exception e){
			throw new RuntimeException(e);
		}
	}
}

jms.JMSMessageQueueReceiver
package jms;
import javax.jms.Destination;
public class JMSMessageQueueReceiver extends JMSMessageConsumer {
	public JMSMessageQueueReceiver(String name) {
		super(name);
	}
	@Override
	public Destination getDestination() {
		return createDefaultDestination(DESTIONATION_TYPE_QUEUE);
	}
}

jms.JMSMessageQueueSender
package jms;
import javax.jms.Destination;
public class JMSMessageQueueSender extends JMSMessageProducer {
	public JMSMessageQueueSender(String name) {
		super(name);
	}
	@Override
	public Destination getDestination() {
		return createDefaultDestination(DESTIONATION_TYPE_QUEUE);
	}
}

jms.JMSMessageTopicPublisher
package jms;
import javax.jms.Destination;
public class JMSMessageTopicPublisher extends JMSMessageProducer {
	public JMSMessageTopicPublisher(String name) {
		super(name);
	}
	@Override
	public Destination getDestination() {
		return createDefaultDestination(DESTIONATION_TYPE_TOPIC);
	}
}

jms.JMSMessageTopicSubscriber
package jms;
import javax.jms.Destination;
public class JMSMessageTopicSubscriber extends JMSMessageConsumer {
	public JMSMessageTopicSubscriber(String name) {
		super(name);
	}
	@Override
	public Destination getDestination() {
		return createDefaultDestination(DESTIONATION_TYPE_TOPIC);
	}
}

jms.JMSUtil
package jms;
import java.text.SimpleDateFormat;
import java.util.*;
import javax.jms.*;
public class JMSUtil {
	private static SimpleDateFormat sdf=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
	@SuppressWarnings("unchecked")
	public static String formatMessage(Message message) {
		StringBuffer result=new StringBuffer();
		try {
			String correlationId=message.getJMSCorrelationID();
			int deliveryMode=message.getJMSDeliveryMode();
			long expiration=message.getJMSExpiration();
			String messageId=message.getJMSMessageID();
			int priority=message.getJMSPriority();
			long timestamp=message.getJMSTimestamp();
			String type=message.getJMSType();
			result.append("##### message property #####\n")
				  .append("correlationId : "+correlationId+"\n")
				  .append("deliveryMode : "+deliveryMode+"\n")
				  .append("expiration : "+expiration+"\n")
				  .append("messageId : "+messageId+"\n")
				  .append("priority : "+priority+"\n")
				  .append("timestamp : "+sdf.format(new Date(timestamp))+"\n")
				  .append("type : "+type+"\n");
			Enumeration<String> names=message.getPropertyNames();
			result.append("##### message content #####\n");
			while(names.hasMoreElements()){
				String name=names.nextElement();
				String value=message.getStringProperty(name);
				result.append(name +" : "+value+"\n");
			}
		} catch (JMSException e) {
			throw new RuntimeException(e);
		}
		return result.toString();
	}
}

jms.Main
package jms;
public class Main {
	public static void main(String[] args) throws InterruptedException {
		sendMesasge();
		Thread.sleep(2000);
		createMessageReceiver();
		
		Thread.sleep(3000);
		System.exit(0);
	}
	/* create message receiver */
	private static void createMessageReceiver(){
		// create queue message receiver
		for(int i=0;i<3;i++){
			Thread t=new Thread(new JMSMessageQueueReceiver("queue_receiver_"+i));
			t.start();
		}
		// create topic message subscriber
		for(int i=0;i<3;i++){
			Thread t=new Thread(new JMSMessageTopicSubscriber("topic_subscriber_"+i));
			t.start();
		}
	}
	/* send message */
	private static void sendMesasge(){
		// create queue message sender
		for(int i=0;i<3;i++){
			Thread t=new Thread(new JMSMessageQueueSender("queue_sender_"+i));
			t.start();
		}
		// create topic publisher
		for(int i=0;i<3;i++){
			Thread t=new Thread(new JMSMessageTopicPublisher("topic_publisher_"+i));
			t.start();
		}
	}
}

JMS Api Demo

  • 0

    开心

    开心

  • 0

    板砖

    板砖

  • 0

    感动

    感动

  • 0

    有用

    有用

  • 0

    疑问

    疑问

  • 0

    难过

    难过

  • 0

    无聊

    无聊

  • 0

    震惊

    震惊

编辑推荐
jms api java消息传递模式有两个api 1、点对点api 2、发布/订阅api 一、点对点消息传送模型api 点对
JMS是Sun公司通过JSR-914项目创建的一种企业消息传送API。JMS自身并不是一种消息传送系统,它是消息
最近温习了下EJB和JMS,整理了下思路,和大家分享下P2P和Pub/Sub的demo ;JBoss 7 集成了HornetQ,J
最近温习了下EJB和JMS,整理了下思路,和大家分享下P2P和Pub/Sub的demo ;JBoss 7 集成了HornetQ,J
最近温习了下EJB和JMS,整理了下思路,和大家分享下P2P和Pub/Sub的demo ;JBoss 7 集成了HornetQ,J
俗话说的好:温故而知新。这就是我发布到这里的原因。 ActiveMQ+Camel+Spring 简单整合 官网我就不列
// 关键字:Porter-Duff package com.example.android.apis.graphics; import com.example.android.
freemarker API demo  FreeMarker 是一款模板引擎:一种基于模板的、用来生成输出文本(任何来自于H
  FreeMarker 是一款模板引擎:一种基于模板的、用来生成输出文本(任何来自于HTML格式的文本用来
1. Forwarding 这个实现很简单,就是启动新的Activity或者Service后,增加一个finish()语句就可以了
版权所有 IT知识库 CopyRight © 2009-2015 IT知识库 IT610.com , All Rights Reserved. 京ICP备09083238号