HornetQ的简单例子

最近对MQ（Message Queue)很感兴趣，准备用到项目上。因为是做Java开发的，所以我找了JMS的MQ。开始看了ActiveMQ和HornetQ。我选择了HornetQ，因为看了一些性能测试，我觉得这款JBoss的MQ相当出色。发现国内关于HornetQ的资料很少，国外有一部分，但是版本都很久了。

自己写了一个例子，环境如下：

1、HornetQ的版本是 hornetq-2.2.5.Final

2、JDK1.6.0_30-b12

3、HornetQ自带example里面的config/stand-alone/non-clustered的配置文件

4、HornetQ作为独立的服务器，运行在一台Dell1950（2CPU，16G内存）上

5、自己写了两个小例子，很简单，就是一个Producer和Consumer

6、发送Person类的实例（必须实现Serializable接口）

7、注意：Producer和Consumer用到的Person类，必须在各个项目的相同的package下面，具有相同的serialVersionUID

8、在classpath下面有jndi.properties文件，里面放置着寻找服务器上面JNDI Server必须的配置

9、在classpath下面有个log4j的配置文件，用来答应日志

代码如下：

 

public class Producer {
	private static Logger logger = Logger.getLogger(Producer.class);

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		try {
			runExample();
		} catch (Exception e) {
			e.printStackTrace();
		}

	}

	private static void runExample() throws NamingException, JMSException {
		InitialContext ic = new InitialContext();
		ConnectionFactory cf = (ConnectionFactory) ic
				.lookup("/ConnectionFactory");
		Queue orderQueue = (Queue) ic.lookup("/queue/ExpiryQueue");
		Connection connection = cf.createConnection();
		Session session = connection.createSession(false,
				Session.AUTO_ACKNOWLEDGE);
		MessageProducer producer = session.createProducer(orderQueue);
		connection.start();
		int count = 0;
		try {
			while (true) {
				Person one = new Person(count++, "xuepeng_" + count);
				ObjectMessage msg = session.createObjectMessage(one);
				producer.send(msg);
				logger.info(Producer.class.getName()
						+ " start to sent message: " + one);
			}
		} finally {
			session.close();
		}
	}

}

 

 

 

public class Person implements Serializable {

	private static final long serialVersionUID = 2670718766927459001L;
	private Integer id;
	private String name;
	private SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
	private String time = format.format(new Date());

	/**
	 * @param id
	 * @param name
	 */
	public Person(Integer id, String name) {
		super();
		this.id = id;
		this.name = name;
	}

	/**
	 * @return the id
	 */
	public Integer getId() {
		return id;
	}

	/**
	 * @param id
	 *            the id to set
	 */
	public void setId(Integer id) {
		this.id = id;
	}

	/**
	 * @return the name
	 */
	public String getName() {
		return name;
	}

	/**
	 * @param name
	 *            the name to set
	 */
	public void setName(String name) {
		this.name = name;
	}



	

	/* (non-Javadoc)
	 * @see java.lang.Object#toString()
	 */
	@Override
	public String toString() {
		return "Person [id=" + id + ", name=" + name + ", time=" + time + "]";
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see java.lang.Object#hashCode()
	 */
	@Override
	public int hashCode() {
		final int prime = 37;
		int result = 17;
		result = prime * result + ((id == null) ? 0 : id.hashCode());
		result = prime * result + ((name == null) ? 0 : name.hashCode());
		return result;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see java.lang.Object#equals(java.lang.Object)
	 */
	@Override
	public boolean equals(Object obj) {
		if (this == obj)
			return true;
		if (obj == null)
			return false;
		if (getClass() != obj.getClass())
			return false;
		Person other = (Person) obj;
		if (id == null) {
			if (other.id != null)
				return false;
		} else if (!id.equals(other.id))
			return false;
		if (name == null) {
			if (other.name != null)
				return false;
		} else if (!name.equals(other.name))
			return false;
		return true;
	}

}

 

 

 

public class Consumer {
	private static Logger logger = Logger.getLogger(Consumer.class);

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		try {
			runExample();
		} catch (Exception e) {
			e.printStackTrace();
		}

	}

	private static void runExample() throws NamingException, JMSException {
		InitialContext ic = new InitialContext();
		ConnectionFactory cf = (ConnectionFactory) ic
				.lookup("/ConnectionFactory");
		Queue orderQueue = (Queue) ic.lookup("/queue/ExpiryQueue");
		Connection connection = cf.createConnection();
		Session session = connection.createSession(false,
				Session.AUTO_ACKNOWLEDGE);
		MessageConsumer consumer = session.createConsumer(orderQueue);
		connection.start();
		try {
			while (true) {
				ObjectMessage messageReceived = (ObjectMessage) consumer.receive();
				org.hornetq.jms.example.Person one = (org.hornetq.jms.example.Person)messageReceived.getObject();
				logger.info(Consumer.class.getName()
						+ " start to receive message: " + one);
			}
		} finally {
			session.close();
		}
	}

}

 启动Linux上面的HorneQ服务之后，运行上面的Producer和Consumer均可以实现通讯。