前一节讲到Petals ESB使用JMS连接ActiveMQ到总线 ，其中因为开发版本的不同和一些细节的配置不到，很难能够正确的使用JMS，如果遇到复杂的JMS需求时，这种方式操作太多，太发太多，而且不容易控制。Petals ESB 4.2支持的要比4.1版本好些。
这一节讲述一下使用JSR181标准的开发方法，完成Petals ESB与ActvieMQ的交互。开发者可能按照这个实现思路来实现多种MQ的接入，或者就叫做Queue Manager.
如下是操作步骤：
一、创建一个JSR181标准的生产者
1）New->Service Provider(SU)->选择JSR181,如图：
2）保存为su-Jsr181-ActiveMq-provide，下一步，如下图：（注意驼峰命名规则）
3）,修改自动生成的ActiveMq.java文件。
import javax.jws.WebMethod;
import javax.jws.WebResult;
import javax.jws.WebService;
/**
* Here is a sample JAX-WS implementation.
* <p>
* For more information about JAX-WS, please visit
* <b>https://jax-ws.dev.java.net/jax-ws-ea3/docs/annotations.html</b>.
* </p>
*
*/
@WebService(serviceName = "ActiveMq", targetNamespace = "http://itownet.com", portName = "ActiveMqPort")
public class ActiveMq {
/**
* Displays "Hello World!".
*/
@WebMethod(operationName = "putMessage")
public void putMessage() {
Sender sender = new Sender();
sender.putMesssage();
}
@WebMethod(operationName = "getMessage")
@WebResult(name = "returnMessage")
public String getMessage() {
Receiver receiver = new Receiver();
return receiver.getMessage();
}
}
4）发送到ActiveMq的示例类Sender.java
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
public class Sender {
private static final int SEND_NUMBER = 5;
public static void putMesssage() {
// ConnectionFactory ：连接工厂，JMS 用它创建连接
ConnectionFactory connectionFactory;
// Connection ：JMS 客户端到JMS Provider 的连接
Connection connection = null;
// Session： 一个发送或接收消息的线程
Session session;
// Destination ：消息的目的地;消息发送给谁.
Destination destination;
// MessageProducer：消息发送者
MessageProducer producer;
// TextMessage message;
// 构造ConnectionFactory实例对象，此处采用ActiveMq的实现jar
connectionFactory = new ActiveMQConnectionFactory(
ActiveMQConnection.DEFAULT_USER,
ActiveMQConnection.DEFAULT_PASSWORD,
"tcp://192.168.21.247:61616");
try {
// 构造从工厂得到连接对象
connection = connectionFactory.createConnection();
// 启动
connection.start();
// 获取操作连接
session = connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);
// 获取session注意参数值xingbo.xu-queue是一个服务器的queue，须在在ActiveMq的console配置
destination = session.createQueue("QM_000000");
// 得到消息生成者【发送者】
producer = session.createProducer(destination);
// 设置不持久化，此处学习，实际根据项目决定
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
// 构造消息，此处写死，项目就是参数，或者方法获取
sendMessage(session, producer);
session.commit();
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
if (null != connection)
connection.close();
} catch (Throwable ignore) {
}
}
}
public static void sendMessage(Session session, MessageProducer producer)
throws Exception {
for (int i = 1; i <= SEND_NUMBER; i++) {
TextMessage message = session
.createTextMessage("ActiveMq 发送的消息" + i);
// 发送消息到目的地方
System.out.println("发送消息：" + "ActiveMq 发送的消息" + i);
producer.send(message);
}
}
}
5)接收消息的实现类Receiver.java
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
public class Receiver {
public static String getMessage() {
// ConnectionFactory ：连接工厂，JMS 用它创建连接
ConnectionFactory connectionFactory;
// Connection ：JMS 客户端到JMS Provider 的连接
Connection connection = null;
// Session： 一个发送或接收消息的线程
Session session;
// Destination ：消息的目的地;消息发送给谁.
Destination destination;
// 消费者，消息接收者
MessageConsumer consumer;
connectionFactory = new ActiveMQConnectionFactory(
ActiveMQConnection.DEFAULT_USER,
ActiveMQConnection.DEFAULT_PASSWORD,
"tcp://192.168.21.247:61616");
try {
// 构造从工厂得到连接对象
connection = connectionFactory.createConnection();
// 启动
connection.start();
// 获取操作连接
session = connection.createSession(Boolean.FALSE,
Session.AUTO_ACKNOWLEDGE);
// 获取session注意参数值xingbo.xu-queue是一个服务器的queue，须在在ActiveMq的console配置
destination = session.createQueue("QM_000000");
consumer = session.createConsumer(destination);
String returnMessage = "";
while (true) {
// 设置接收者接收消息的时间，为了便于测试，这里谁定为100s
TextMessage message = (TextMessage) consumer.receive(100000);
if (null != message) {
System.out.println("收到消息" + message.getText());
returnMessage += ";" + message.getText();
} else {
break;
}
}
return returnMessage;
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
if (null != connection)
connection.close();
} catch (Throwable ignore) {
}
}
return "无消息!";
}
}
注意，要把activeMQ相关的jar包导入到项目当中。也就是Build Path里。
最后，生成通过Petals->Fast Export to Petals,生成zip文件，先检查一下ZIP文件下有没有ActiveMq相关的jar包。如果没有，请手动把ActiveMq相关的包放在ZIP里，然后部署在Petals ESB当中。
二、JSR181的生产者生成WSDL服务
操作如下：
三、创建一个消费者
1）New->Service Consumer(SU)，选择刚才的SOAP，本文想借消费者的调用放入到ActiveMqj里消费或者取回消息。
2）把项目保存为su-SOAP-ActiveMq-consume。
最后，通过Petals->Fast Export to Petals,生成zip文件，部署在Petals ESB当中。
通过观察Petals ESB的容器的部署情况，会发现生成一个Web service在ESB当中，比如：
http://192.168.21.187:8084/petals/services/ActiveMq?wsdl
四、使用soapUI完成测试。
创建soapUI的项目，并打开activeMq的控制台，观察结果。看看是否预期。
注意图示当中的观察点。
如上只为开发者提供JMS客户端调用的方法完成的实现，好处是可控性好，适合复杂业务逻辑。