? ? ? ?前一节讲到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); // 发送消息到目的地方