ActiveMQ JMS的测试

有二种方式可以测试。

1. 透过testng, 在测试类中完成的jms服务器的启动与关闭。

a.初始化jms服务器

@BeforeClass(groups = "jms")
	    public void setupActiveMQ() throws Exception {
	        BrokerService broker = new BrokerService();
	        broker.setPersistent(false);
	        broker.setUseJmx(false);
	        broker.start();
	        URI uri = broker.getVmConnectorURI();
	        ConnectionFactory factory = new ActiveMQConnectionFactory(uri);
	        //String uri = "vm://localhost?broker.persistent=false";
	        //ConnectionFactory factory = new ActiveMQConnectionFactory(uri);
	        Connection connection = factory.createConnection();
	        connection.start();
	        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
	        destination = session.createQueue("TestQueue@router1");
	    }

 b. 运行测试类

@Test(groups = "jms")
	    public void sendMessage() throws JMSException {
	        TextMessage msg = session.createTextMessage();
	        msg.setText("hello");
	        session.createProducer(destination).send(msg);
	    }

c. 验证接收到的消息

 @Test(groups = "jms", dependsOnMethods = "sendMessage", timeOut = 1000)
	    public void receiveMessage() throws JMSException {
	        MessageConsumer consumer = session.createConsumer(destination);
	        TextMessage msg = (TextMessage) consumer.receive();
	        assert "hello".equals(msg.getText());
	    }

 

2. 透过maven-activemq-plugin插件，运行jms服务器，然后分别实现发送与接收

a. maven 配置

<plugin>
				<groupId>org.apache.activemq.tooling</groupId>
				<artifactId>maven-activemq-plugin</artifactId>
				<version>5.2.0</version>
				<configuration>
					<configUri>
						xbean:file:src/main/resources/net/sunbeta/jms/activemq.xml
					</configUri>
					<fork>false</fork>
					<systemProperties>
						<property>
							<name>javax.net.ssl.keyStorePassword</name>
							<value>password</value>
						</property>
						<property>
							<name>org.apache.activemq.default.directory.prefix</name>
							<value>./target/</value>
						</property>
					</systemProperties>
				</configuration>
				<dependencies>
					<dependency>
						<groupId>org.springframework</groupId>
						<artifactId>spring</artifactId>
						<version>2.5.6</version>
					</dependency>
					<dependency>
						<groupId>org.mortbay.jetty</groupId>
						<artifactId>jetty-xbean</artifactId>
						<version>6.1.11</version>
					</dependency>
					<dependency>
						<groupId>org.apache.camel</groupId>
						<artifactId>camel-activemq</artifactId>
						<version>1.1.0</version>
					</dependency>
				</dependencies>
			</plugin>

 其中 activemq.xml 的配置为:

<?xml version="1.0"?>
<beans xmlns="http://www.springframework.org/schema/beans"
  xmlns:amq="http://activemq.apache.org/schema/core"
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://www.springframework.org/schema/beans
  http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
  http://activemq.apache.org/schema/core
  http://activemq.apache.org/schema/core/activemq-core.xsd">
  
  <broker xmlns="http://activemq.apache.org/schema/core" brokerName="localhost" dataDirectory="./data">
    <!-- The transport connectors ActiveMQ will listen to -->
    <transportConnectors>
      <transportConnector name="openwire" uri="tcp://localhost:61616"/>
    </transportConnectors>
  </broker>
</beans>

 b.消息发送

public class JmsProducer {

	private ConnectionFactory factory;
	private Connection connection;
	private Session session;
	private MessageProducer producer;

	public JmsProducer(ConnectionFactory factory, String queueName)
			throws JMSException {
		this.factory = factory;
		connection = factory.createConnection();
		connection.start();
		session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
		Destination destination = session.createQueue(queueName);
		producer = session.createProducer(destination);
	}

	public void run() throws JMSException {
		for (int i = 0; i < 100; i++) {
			System.out.println("Creating Message " + i);
			Message message = session.createTextMessage("Hello World!");
			producer.send(message);
		}
	}

	public void close() throws JMSException {
		if (connection != null) {
			connection.close();
		}
	}
}

 创建一个broker去调用它

public class JmsBroker {

	public static String brokerURL = "tcp://localhost:61616";

	// mvn clean compile exec:java
	// -Dexec.mainClass=net.sunbeta.test.jms.JmsBroker
	public static void main(String[] args) throws Exception {
		// setup the connection to ActiveMQ
		ConnectionFactory factory = new ActiveMQConnectionFactory(brokerURL);

		JmsProducer producer = new JmsProducer(factory, "test");
		producer.run();
		producer.close();
	}
}

 c. 消息接收

public class JmsConsumer implements MessageListener{
	public static String brokerURL = "tcp://localhost:61616";

	private ConnectionFactory factory;
	private Connection connection;
	private Session session;
	private MessageConsumer consumer;

	public static void main(String[] args) {
		JmsConsumer app = new JmsConsumer();
		app.run();
	}

	public void run() {
		try {
			ConnectionFactory factory = new ActiveMQConnectionFactory(brokerURL);
			connection = factory.createConnection();
			connection.start();
			session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
			Destination destination = session.createQueue("test");
			consumer = session.createConsumer(destination);
			consumer.setMessageListener(this);
		} catch (Exception e) {
			System.out.println("Caught:" + e);
			e.printStackTrace();
		}
	}

	public void onMessage(Message message) {
		try {
			if (message instanceof TextMessage) {
				TextMessage txtMessage = (TextMessage) message;
				System.out.println("Message received: " + txtMessage.getText());
			} else {
				System.out.println("Invalid message received.");
			}
		} catch (JMSException e) {
			System.out.println("Caught:" + e);
			e.printStackTrace();
		}
	}

 

OK !