EJB3 学习笔记七

开发消息驱动bean

Java消息服务 java message service用于访问企业消息系统的开发商中立的api

应用场合:两个没有直接代码关联的应用a和b 可以通过消息服务器进行消息的传递

消息由三部分组成:

头(header)

属性(property)

主体(body)

消息的几中常见类型

streamMessage

mapMessage

textMessage

objectMessage

bytesMessage

消息的传递方式有两种:

点对点(point topoint) 一条消息只能有一个接收方

发布定阅(publish/subscribe)一条消息有多个接收方

修改配置文件standlone.xml

<?xml version='1.0'encoding='UTF-8'?>

<serverxmlns="urn:jboss:domain:1.2">

<extensionmodule="org.jboss.as.clustering.infinispan"/>

<extensionmodule="org.jboss.as.cmp"/>

<extensionmodule="org.jboss.as.configadmin"/>

<extensionmodule="org.jboss.as.connector"/>

<extensionmodule="org.jboss.as.deployment-scanner"/>

<extensionmodule="org.jboss.as.ee"/>

<extensionmodule="org.jboss.as.ejb3"/>

<extensionmodule="org.jboss.as.jacorb"/>

<extensionmodule="org.jboss.as.jaxr"/>

<extensionmodule="org.jboss.as.jaxrs"/>

<extensionmodule="org.jboss.as.jdr"/>

<extensionmodule="org.jboss.as.jmx"/>

<extensionmodule="org.jboss.as.jsr77"/>

<extensionmodule="org.jboss.as.logging"/>

<extensionmodule="org.jboss.as.mail"/>

<extensionmodule="org.jboss.as.messaging"/>

<extensionmodule="org.jboss.as.modcluster"/>

<extensionmodule="org.jboss.as.osgi"/>

<extensionmodule="org.jboss.as.pojo"/>

<extensionmodule="org.jboss.as.remoting"/>

<extensionmodule="org.jboss.as.sar"/>

<extensionmodule="org.jboss.as.security"/>

<extensionmodule="org.jboss.as.transactions"/>

<extensionmodule="org.jboss.as.web"/>

<extensionmodule="org.jboss.as.webservices"/>

<extensionmodule="org.jboss.as.weld"/>

</extensions>

<security-realms>

<security-realmname="ManagementRealm">

<propertiespath="mgmt-users.properties"relative-to="jboss.server.config.dir"/>

</authentication>

</security-realm>

<security-realmname="ApplicationRealm">

</authentication>

</security-realm>

</security-realms>

<management-interfaces>

<native-interfacesecurity-realm="ManagementRealm">

<socket-bindingnative="management-native"/>

</native-interface>

<http-interfacesecurity-realm="ManagementRealm">

<socket-bindinghttp="management-http"/>

</http-interface>

</management-interfaces>

</management>

<subsystemxmlns="urn:jboss:domain:logging:1.1">

<console-handlername="CONSOLE">

<pattern-formatterpattern="%d{HH:mm:ss,SSS} %-5p [%c] (%t) %s%E%n"/>

</formatter>

</console-handler>

<periodic-rotating-file-handlername="FILE">

<pattern-formatterpattern="%d{HH:mm:ss,SSS} %-5p [%c] (%t) %s%E%n"/>

</formatter>

<filerelative-to="jboss.server.log.dir" path="server.log"/>

<suffixvalue=".yyyy-MM-dd"/>

</periodic-rotating-file-handler>

<loggercategory="com.arjuna">

</logger>

<loggercategory="org.apache.tomcat.util.modeler">

</logger>

<loggercategory="sun.rmi">

</logger>

<loggercategory="jacorb">

</logger>

<loggercategory="jacorb.config">

</logger>

<root-logger>

<handlername="CONSOLE"/>

</handlers>

</root-logger>

</subsystem>

<subsystemxmlns="urn:jboss:domain:cmp:1.0"/>

<subsystemxmlns="urn:jboss:domain:configadmin:1.0"/>

<subsystemxmlns="urn:jboss:domain:datasources:1.0">

<datasourcejndi-name="java:jboss/datasources/ExampleDS"pool-name="ExampleDS" enabled="true"use-java-context="true">

<connection-url>jdbc:h2:mem:test;DB_CLOSE_DELAY=-1</connection-url>

<user-name>sa</user-name>

</security>

</datasource>

<datasourcejndi-name="java:/MySqlDS" pool-name="MySqlDS"enabled="true" use-java-context="true">

<connection-url>jdbc:mysql://localhost:3306/shop</connection-url>

<driver>mysql</driver>

<user-name>root</user-name>

</security>

</datasource>

<datasourcejndi-name="java:/OracleDS" pool-name="OracleDS"enabled="true" use-java-context="true">

<connection-url>jdbc:oracle:thin:@localhost:1521:RETACN</connection-url>

<driver>oracle</driver>

<user-name>bb</user-name>

</security>

</datasource>

<xa-datasource-class>org.h2.jdbcx.JdbcDataSource</xa-datasource-class>

</driver>

<driver-class>org.gjt.mm.mysql.Driver</driver-class>

<xa-datasource-class>com.mysql.jdbc.jdbc2.optional.MysqlXADataSource</xa-datasource-class>

</driver>

<driver-class>oracle.jdbc.driver.OracleDriver</driver-class>

<xa-datasource-class>oracle.jdbc.xa.client.OracleXADataSource</xa-datasource-class>

</driver>

</drivers>

</datasources>

</subsystem>

<subsystemxmlns="urn:jboss:domain:deployment-scanner:1.1">

<deployment-scannerpath="deployments" relative-to="jboss.server.base.dir"scan-interval="5000"/>

</subsystem>

<subsystemxmlns="urn:jboss:domain:ee:1.0"/>

<subsystemxmlns="urn:jboss:domain:ejb3:1.2">

<session-bean>

<bean-instance-pool-refpool-name="slsb-strict-max-pool"/>

</stateless>

<statefuldefault-access-timeout="5000" cache-ref="simple"clustered-cache-ref="clustered"/>

<singletondefault-access-timeout="5000"/>

</session-bean>

<mdb>

<resource-adapter-ref resource-adapter-name="hornetq-ra"/>

<bean-instance-pool-refpool-name="mdb-strict-max-pool"/>

</mdb>

<pools>

<bean-instance-pools>

<strict-max-poolname="slsb-strict-max-pool" max-pool-size="20"instance-acquisition-timeout="5" instance-acquisition-timeout-unit="MINUTES"/>

<strict-max-poolname="mdb-strict-max-pool" max-pool-size="20"instance-acquisition-timeout="5"instance-acquisition-timeout-unit="MINUTES"/>

</bean-instance-pools>

</pools>

</caches>

<passivation-stores>

<file-passivation-storename="file"/>

<cluster-passivation-storename="infinispan" cache-container="ejb"/>

</passivation-stores>

<asyncthread-pool-name="default"/>

<timer-servicethread-pool-name="default">

<data-storepath="timer-service-data"relative-to="jboss.server.data.dir"/>

</timer-service>

<remoteconnector-ref="remoting-connector"thread-pool-name="default"/>

<thread-pools>

<thread-poolname="default">

<max-threadscount="10"/>

<keepalive-timetime="100" unit="milliseconds"/>

</thread-pool>

</thread-pools>

</subsystem>

<subsystemxmlns="urn:jboss:domain:infinispan:1.2"default-cache-container="cluster">

<cache-containername="cluster" aliases="ha-partition"default-cache="default">

<transportlock-timeout="60000"/>

<replicated-cachename="default" mode="SYNC" batching="true">

<lockingisolation="REPEATABLE_READ"/>

</replicated-cache>

</cache-container>

<cache-containername="web" aliases="standard-session-cache"default-cache="repl">

<transportlock-timeout="60000"/>

<replicated-cachename="repl" mode="ASYNC" batching="true">

<file-store/>

</replicated-cache>

<replicated-cachename="sso" mode="SYNC" batching="true"/>

<distributed-cachename="dist" mode="ASYNC" batching="true">

<file-store/>

</distributed-cache>

</cache-container>

<cache-containername="ejb" aliases="sfsb sfsb-cache"default-cache="repl">

<transportlock-timeout="60000"/>

<replicated-cachename="repl" mode="ASYNC" batching="true">

<evictionstrategy="LRU"/>

<file-store/>

</replicated-cache>

<replicated-cachename="remote-connector-client-mappings" mode="SYNC"batching="true"/>

<distributed-cachename="dist" mode="ASYNC" batching="true">

<evictionstrategy="LRU"/>

<file-store/>

</distributed-cache>

</cache-container>

<cache-containername="hibernate" default-cache="local-query">

<transportlock-timeout="60000"/>

<local-cachename="local-query">

<transactionmode="NONE"/>

<expirationmax-idle="100000"/>

</local-cache>

<invalidation-cachename="entity" mode="SYNC">

<transactionmode="NON_XA"/>

<expirationmax-idle="100000"/>

</invalidation-cache>

<replicated-cachename="timestamps" mode="ASYNC">

<transactionmode="NONE"/>

<evictionstrategy="NONE"/>

</replicated-cache>

</cache-container>

</subsystem>

<subsystemxmlns="urn:jboss:domain:jacorb:1.1">

<orb>

<initializerssecurity="on" transactions="spec"/>

</orb>

</subsystem>

<subsystemxmlns="urn:jboss:domain:jaxr:1.1">

<connection-factoryjndi-name="java:jboss/jaxr/ConnectionFactory"/>

</subsystem>

<subsystemxmlns="urn:jboss:domain:jaxrs:1.0"/>

<subsystemxmlns="urn:jboss:domain:jca:1.1">

<archive-validationenabled="true" fail-on-error="true"fail-on-warn="false"/>

<bean-validation enabled="true"/>

<default-workmanager>

<short-running-threads>

<core-threadscount="50"/>

<queue-lengthcount="50"/>

<max-threadscount="50"/>

<keepalive-time time="10"unit="seconds"/>

</short-running-threads>

<long-running-threads>

<core-threadscount="50"/>

<queue-lengthcount="50"/>

<max-threadscount="50"/>

<keepalive-time time="10"unit="seconds"/>

</long-running-threads>

</default-workmanager>

<cached-connection-manager/>

</subsystem>

<subsystemxmlns="urn:jboss:domain:jdr:1.0"/>

<protocoltype="PING"/>

<protocoltype="MERGE2"/>

<protocoltype="VERIFY_SUSPECT"/>

<protocoltype="BARRIER"/>

<protocoltype="pbcast.NAKACK"/>

<protocoltype="UNICAST2"/>

<protocoltype="pbcast.STABLE"/>

<protocoltype="pbcast.GMS"/>

<protocoltype="FRAG2"/>

</stack>

<protocoltype="VERIFY_SUSPECT"/>

<protocoltype="BARRIER"/>

<protocoltype="pbcast.NAKACK"/>

<protocoltype="UNICAST2"/>

<protocoltype="pbcast.STABLE"/>

<protocoltype="FRAG2"/>

</stack>

</subsystem>

<subsystemxmlns="urn:jboss:domain:jmx:1.1">

<show-modelvalue="true"/>

<remoting-connector/>

</subsystem>

<subsystemxmlns="urn:jboss:domain:jpa:1.0">

<jpadefault-datasource=""/>

</subsystem>

<subsystemxmlns="urn:jboss:domain:jsr77:1.0"/>

<subsystemxmlns="urn:jboss:domain:mail:1.0">

<mail-sessionjndi-name="java:jboss/mail/Default">

<smtp-serveroutbound-socket-binding-ref="mail-smtp"/>

</mail-session>

</subsystem>

<subsystemxmlns="urn:jboss:domain:messaging:1.1">

<hornetq-server>

<persistence-enabled>true</persistence-enabled>

<journal-file-size>102400</journal-file-size>

<journal-min-files>2</journal-min-files>

<netty-connectorname="netty" socket-binding="messaging"/>

<netty-connectorname="netty-throughput"socket-binding="messaging-throughput">

</netty-connector>

<in-vm-connectorname="in-vm" server-id="0"/>

</connectors>

<netty-acceptorname="netty" socket-binding="messaging"/>

<netty-acceptorname="netty-throughput"socket-binding="messaging-throughput">

<paramkey="direct-deliver" value="false"/>

</netty-acceptor>

<in-vm-acceptorname="in-vm" server-id="0"/>

</acceptors>

<broadcast-groups>

<broadcast-groupname="bg-group1">

<group-address>231.7.7.7</group-address>

<group-port>9876</group-port>

<broadcast-period>5000</broadcast-period>

<connector-ref>

netty

</connector-ref>

</broadcast-group>

</broadcast-groups>

<discovery-groups>

<discovery-groupname="dg-group1">

<group-address>231.7.7.7</group-address>

<group-port>9876</group-port>

<refresh-timeout>10000</refresh-timeout>

</discovery-group>

</discovery-groups>

<cluster-connections>

<cluster-connectionname="my-cluster">

<connector-ref>netty</connector-ref>

<discovery-group-refdiscovery-group-name="dg-group1"/>

</cluster-connection>

</cluster-connections>

<security-settings>

<security-settingmatch="#">

<permissiontype="consume" roles="guest"/>

<permissiontype="createNonDurableQueue" roles="guest"/>

<permissiontype="deleteNonDurableQueue" roles="guest"/>

</security-setting>

</security-settings>

<address-settings>

<address-settingmatch="#">

<dead-letter-address>jms.queue.DLQ</dead-letter-address>

<expiry-address>jms.queue.ExpiryQueue</expiry-address>

<redelivery-delay>0</redelivery-delay>

<max-size-bytes>10485760</max-size-bytes>

<address-full-policy>BLOCK</address-full-policy>

<message-counter-history-day-limit>10</message-counter-history-day-limit>

<redistribution-delay>1000</redistribution-delay>

</address-setting>

</address-settings>

<jms-connection-factories>

<connection-factoryname="InVmConnectionFactory">

<connector-refconnector-name="in-vm"/>

</connectors>

<entryname="java:/ConnectionFactory"/>

</entries>

</connection-factory>

<connection-factoryname="RemoteConnectionFactory">

<connector-refconnector-name="netty"/>

</connectors>

<entryname="RemoteConnectionFactory"/>

<entryname="java:jboss/exported/jms/RemoteConnectionFactory"/>

</entries>

</connection-factory>

<pooled-connection-factoryname="hornetq-ra">

<transactionmode="xa"/>

<connector-refconnector-name="in-vm"/>

</connectors>

<entryname="java:/JmsXA"/>

</entries>

</pooled-connection-factory>

</jms-connection-factories>

<jms-destinations>

<jms-queue name="testQueue">

<entryname="queue/test"/>

<entryname="java:jboss/exported/jms/queue/test"/>

</jms-queue>

<jms-topicname="testTopic">

<entryname="topic/test"/>

<entryname="java:jboss/exported/jms/topic/test"/>

</jms-topic>

</jms-destinations>

</hornetq-server>

</subsystem>

<subsystemxmlns="urn:jboss:domain:modcluster:1.0">

<mod-cluster-configadvertise-socket="modcluster">

<dynamic-load-provider>

<load-metrictype="busyness"/>

</dynamic-load-provider>

</mod-cluster-config>

</subsystem>

<subsystemxmlns="urn:jboss:domain:osgi:1.2" activation="lazy">

<propertyname="org.osgi.framework.startlevel.beginning">

</property>

</properties>

<capabilityname="javax.transaction.api"/>

<capabilityname="org.apache.felix.log" startlevel="1"/>

<capabilityname="org.jboss.osgi.logging" startlevel="1"/>

<capabilityname="org.apache.felix.configadmin" startlevel="1"/>

<capabilityname="org.jboss.as.osgi.configadmin" startlevel="1"/>

</capabilities>

</subsystem>

<subsystemxmlns="urn:jboss:domain:pojo:1.0"/>

<subsystemxmlns="urn:jboss:domain:remoting:1.1">

<connectorname="remoting-connector" socket-binding="remoting" security-realm="ApplicationRealm"/>

</subsystem>

<subsystemxmlns="urn:jboss:domain:resource-adapters:1.0"/>

<subsystemxmlns="urn:jboss:domain:sar:1.0"/>

<subsystemxmlns="urn:jboss:domain:security:1.1">

<security-domains>

<security-domainname="other" cache-type="default">

<login-modulecode="Remoting" flag="optional">

<module-optionname="password-stacking" value="useFirstPass"/>

</login-module>

<login-modulecode="RealmUsersRoles" flag="required">

<module-optionname="usersProperties" value="${jboss.server.config.dir}/application-users.properties"/>

<module-optionname="rolesProperties"value="${jboss.server.config.dir}/application-roles.properties"/>

<module-optionname="realm" value="ApplicationRealm"/>

<module-option name="password-stacking"value="useFirstPass"/>

</login-module>

</authentication>

</security-domain>

<security-domainname="jboss-web-policy" cache-type="default">

<policy-modulecode="Delegating" flag="required"/>

</authorization>

</security-domain>

<security-domainname="jboss-ejb-policy" cache-type="default">

<policy-modulecode="Delegating" flag="required"/>

</authorization>

</security-domain>

</security-domains>

</subsystem>

<subsystemxmlns="urn:jboss:domain:threads:1.1"/>

<subsystemxmlns="urn:jboss:domain:transactions:1.1">

<core-environment>

<process-id>

<uuid/>

</process-id>

</core-environment>

<recovery-environmentsocket-binding="txn-recovery-environment"status-socket-binding="txn-status-manager"/>

<coordinator-environmentdefault-timeout="300"/>

</subsystem>

<subsystemxmlns="urn:jboss:domain:web:1.1"default-virtual-server="default-host" native="false">

<virtual-servername="default-host" enable-welcome-root="true">

<aliasname="localhost"/>

<aliasname="example.com"/>

</virtual-server>

</subsystem>

<subsystemxmlns="urn:jboss:domain:webservices:1.1">

<modify-wsdl-address>true</modify-wsdl-address>

<wsdl-host>${jboss.bind.address:127.0.0.1}</wsdl-host>

<endpoint-configname="Standard-Endpoint-Config"/>

<endpoint-configname="Recording-Endpoint-Config">

<pre-handler-chainname="recording-handlers" protocol-bindings="##SOAP11_HTTP##SOAP11_HTTP_MTOM ##SOAP12_HTTP ##SOAP12_HTTP_MTOM">

<handlername="RecordingHandler"class="org.jboss.ws.common.invocation.RecordingServerHandler"/>

</pre-handler-chain>

</endpoint-config>

</subsystem>

</profile>

<interfacename="management">

<inet-addressvalue="${jboss.bind.address.management:127.0.0.1}"/>

</interface>

<interfacename="public">

<inet-addressvalue="${jboss.bind.address:127.0.0.1}"/>

</interface>

<interfacename="unsecure">

<inet-addressvalue="${jboss.bind.address.unsecure:127.0.0.1}"/>

</interface>

</interfaces>

<socket-binding-groupname="standard-sockets" default-interface="public"port-offset="${jboss.socket.binding.port-offset:0}">

<socket-bindingname="management-native" interface="management"port="${jboss.management.native.port:9999}"/>

<socket-bindingname="management-http" interface="management"port="${jboss.management.http.port:9990}"/>

<socket-bindingname="management-https" interface="management"port="${jboss.management.https.port:9443}"/>

<socket-bindingname="ajp" port="8009"/>

<socket-bindingname="http" port="8080"/>

<socket-bindingname="https" port="8443"/>

<socket-bindingname="jacorb" interface="unsecure" port="3528"/>

<socket-bindingname="jacorb-ssl" interface="unsecure"port="3529"/>

<socket-bindingname="jgroups-diagnostics" port="0"multicast-address="224.0.75.75" multicast-port="7500"/>

<socket-bindingname="jgroups-mping" port="0"multicast-address="${jboss.default.multicast.address:230.0.0.4}"multicast-port="45700"/>

<socket-bindingname="jgroups-tcp" port="7600"/>

<socket-bindingname="jgroups-tcp-fd" port="57600"/>

<socket-bindingname="jgroups-udp" port="55200"multicast-address="${jboss.default.multicast.address:230.0.0.4}"multicast-port="45688"/>

<socket-bindingname="jgroups-udp-fd" port="54200"/>

<socket-bindingname="messaging" port="5445"/>

<socket-bindingname="messaging-throughput" port="5455"/>

<socket-bindingname="modcluster" port="0" multicast-address="224.0.1.105"multicast-port="23364"/>

<socket-bindingname="osgi-http" interface="management"port="8090"/>

<socket-bindingname="remoting" port="4447"/>

<socket-bindingname="txn-recovery-environment" port="4712"/>

<socket-bindingname="txn-status-manager" port="4713"/>

<outbound-socket-bindingname="mail-smtp">

<remote-destinationhost="localhost" port="25"/>

</outbound-socket-binding>

</socket-binding-group>

</server>

点对点模式,示例代码如下:

/**

* FileName:JmsMessProducer.java

* Author:<ahref="mailto:zhenhuayue@sina.com">Retacn</a>

* CreateTime: 2016-7-24

// Package Information

package cn.yue.ejb.drivenbean;

/**

* 消息发送接口基类

* @version

* @Description:

* @author <ahref="mailto:zhenhuayue@sina.com">Retacn</a>

* @since 2016-7-24

public interface JmsMessProducer {

/**

* 发送消息

* @param message 消息内容

public void send(String message);

}

/**

* FileName:JmsMessP2PProducer.java

* Author:<a href="mailto:zhenhuayue@sina.com">Retacn</a>

* CreateTime: 2016-7-24

// Package Information

package cn.yue.ejb.drivenbean;

/**

* 点对点消息发送接口

* @version

* @Description:

* @author <ahref="mailto:zhenhuayue@sina.com">Retacn</a>

* @since 2016-7-24

public interface JmsMessP2PProducerextends JmsMessProducer {

}

/**

* FileName:JmsMessP2PProducterBean.java

* Author:<ahref="mailto:zhenhuayue@sina.com">Retacn</a>

* CreateTime: 2016-7-24

// Package Information

package cn.yue.ejb.drivenbean;

import javax.annotation.Resource;

import javax.ejb.Remote;

import javax.ejb.Stateless;

import javax.jms.Destination;

import javax.jms.JMSException;

import javax.jms.MessageProducer;

import javax.jms.QueueConnection;

import javax.jms.QueueConnectionFactory;

import javax.jms.QueueSession;

import javax.jms.TextMessage;

/**

* 点对点模式消息驱动bean(实现消息发送)

* @version

* @Description:

* @author <ahref="mailto:zhenhuayue@sina.com">Retacn</a>

* @since 2016-7-24

@Stateless

@Remote(JmsMessP2PProducer.class)

public class JmsMessP2PProducerBeanimplements JmsMessP2PProducer {

/** 注入链接工厂 **/

@Resource(mappedName = "java:/ConnectionFactory")

private QueueConnectionFactory conFactory;

/** 注入目标地址 **/

@Resource(mappedName = "java:/queue/test")

private Destination destination;

@Override

public void send(String message) {

QueueConnection queueConnection = null;

QueueSession session = null;

try {

// 创建链接

queueConnection =conFactory.createQueueConnection();

// 消息接收到后的确认模式为自动确认

session =queueConnection.createQueueSession(false, QueueSession.AUTO_ACKNOWLEDGE);

// 消息提供者

MessageProducer producer =session.createProducer(destination);

// 文本消息

TextMessage tMessage =session.createTextMessage(message);

// 发送消息

producer.send(tMessage);

System.out.println(JmsMessP2PProducerBean.class.getSimpleName() + ": 发送的消息为: "

+ tMessage.getText());

} catch (JMSException e) {

e.printStackTrace();

} finally {

try {

session.close();

queueConnection.close();

} catch (JMSException e) {

e.printStackTrace();

}

/**

* FileName:JmsMessP2PConsumer.java

* Author:<ahref="mailto:zhenhuayue@sina.com">Retacn</a>

* CreateTime: 2016-7-24

// Package Information

package cn.yue.ejb.drivenbean;

importjavax.ejb.ActivationConfigProperty;

import javax.ejb.MessageDriven;

import javax.jms.JMSException;

import javax.jms.Message;

import javax.jms.MessageListener;

import javax.jms.TextMessage;

/**

* 点对点消息接收者

* @version

* @Description:

* @author <ahref="mailto:zhenhuayue@sina.com">Retacn</a>

* @since 2016-7-24

@MessageDriven(activationConfig = {

@ActivationConfigProperty(propertyName = "destinationType",propertyValue = "javax.jms.Queue"),

@ActivationConfigProperty(propertyName = "destination",propertyValue = "java:/queue/test")

})

public class JmsMessP2PConsumerimplements MessageListener {

@Override

public void onMessage(Message msg) {

TextMessage tMsg = (TextMessage) msg;

try {

System.out.println(JmsMessP2PConsumer.class.getSimpleName() + ": 接收到的消息为: "

+ tMsg.getText());

} catch (JMSException e) {

e.printStackTrace();

}

发布定阅模式,示例代码如下:

/**

* FileName:JmsMessTopicProducer.java

* Author:<ahref="mailto:zhenhuayue@sina.com">Retacn</a>

* CreateTime: 2016-7-24

// Package Information

package cn.yue.ejb.drivenbean;

/**

* 消息发布接口(发布定阅模式)

* @version

* @Description:

* @author <ahref="mailto:zhenhuayue@sina.com">Retacn</a>

* @since 2016-7-24

public interface JmsMessTopicProducerextends JmsMessProducer {

}

/**

* FileName:JmsMessTopicProducerBean.java

* Author:<ahref="mailto:zhenhuayue@sina.com">Retacn</a>

* CreateTime: 2016-7-24

// Package Information

package cn.yue.ejb.drivenbean;

import javax.annotation.Resource;

import javax.ejb.Remote;

import javax.ejb.Stateless;

import javax.jms.Destination;

import javax.jms.JMSException;

import javax.jms.MessageProducer;

import javax.jms.TextMessage;

import javax.jms.TopicConnection;

importjavax.jms.TopicConnectionFactory;

import javax.jms.TopicSession;

/**

* 发布定阅模式消息驱动bean

* @version

* @Description:

* @author <ahref="mailto:zhenhuayue@sina.com">Retacn</a>

* @since 2016-7-24

@Stateless

@Remote(JmsMessTopicProducer.class)

public class JmsMessTopicProducerBeanimplements JmsMessTopicProducer {

/** 注入链接工厂 **/

@Resource(mappedName = "java:/ConnectionFactory")

private TopicConnectionFactory conFactory;

/** 注入目标地址 **/

@Resource(mappedName = "java:/topic/test")

private Destination destination;

@Override

public void send(String message) {

TopicConnection connection = null;

TopicSession session = null;

try {

connection =conFactory.createTopicConnection();

session =connection.createTopicSession(false, TopicSession.AUTO_ACKNOWLEDGE);

MessageProducer producer =session.createProducer(destination);

TextMessage tMsg =session.createTextMessage(message);

producer.send(tMsg);

System.out.println(JmsMessTopicProducerBean.class.getSimpleName() + " 发送的消息为: "

+ tMsg.getText());

} catch (JMSException e) {

e.printStackTrace();

} finally {

try {

session.close();

connection.close();

} catch (JMSException e) {

e.printStackTrace();

}

/**

* FileName:JmsMessTopicConsumerOne.java

* Author:<ahref="mailto:zhenhuayue@sina.com">Retacn</a>

* CreateTime: 2016-7-24

// Package Information

package cn.yue.ejb.drivenbean;

importjavax.ejb.ActivationConfigProperty;

import javax.ejb.MessageDriven;

import javax.jms.JMSException;

import javax.jms.Message;

import javax.jms.MessageListener;

import javax.jms.TextMessage;

/**

* 发布定阅模式, 消息接收者一

* @version

* @Description:

* @author <ahref="mailto:zhenhuayue@sina.com">Retacn</a>

* @since 2016-7-24

@MessageDriven(activationConfig = {

@ActivationConfigProperty(propertyName = "destinationType",propertyValue = "javax.jms.Topic"),

@ActivationConfigProperty(propertyName = "destination",propertyValue = "java:/topic/test")

})

public class JmsMessTopicConsumerOneimplements MessageListener {

@Override

public void onMessage(Message msg) {

TextMessage message = (TextMessage) msg;

try {

System.out.println(JmsMessTopicConsumerOne.class.getSimpleName() +" 接收到的消息为: "

+ message.getText());

} catch (JMSException e) {

e.printStackTrace();

}

/**

* FileName:JmsMessTopicConsumerTwo.java

* Author:<ahref="mailto:zhenhuayue@sina.com">Retacn</a>

* CreateTime: 2016-7-24

// Package Information

package cn.yue.ejb.drivenbean;

importjavax.ejb.ActivationConfigProperty;

import javax.ejb.MessageDriven;

import javax.jms.JMSException;

import javax.jms.Message;

import javax.jms.MessageListener;

import javax.jms.TextMessage;

/**

* 发布定阅模式,消息接收者二

* @version

* @Description:

* @author <ahref="mailto:zhenhuayue@sina.com">Retacn</a>

* @since 2016-7-24

@MessageDriven(activationConfig = {

@ActivationConfigProperty(propertyName = "destinationType",propertyValue = "javax.jms.Topic"),

@ActivationConfigProperty(propertyName = "destination",propertyValue = "java:/topic/test")

})

public class JmsMessTopicConsumerTwoimplements MessageListener {

@Override

public void onMessage(Message msg) {

TextMessage message = (TextMessage) msg;

try {

System.out.println(JmsMessTopicConsumerTwo.class.getSimpleName() +" 接收到的消息为: "

+ message.getText());

} catch (JMSException e) {

e.printStackTrace();

}

客户端测试代码如下:

/**

* FileName:DrivenBeanClient.java

* Author:<ahref="mailto:zhenhuayue@sina.com">Retacn</a>

* CreateTime: 2016-7-24

// Package Information

package cn.yue.ejb.test;

importcn.yue.ejb.drivenbean.JmsMessP2PProducer;

importcn.yue.ejb.drivenbean.JmsMessProducer;

importcn.yue.ejb.drivenbean.JmsMessTopicProducer;

import java.util.Hashtable;

import javax.naming.Context;

import javax.naming.InitialContext;

import javax.naming.NamingException;

/**

* 测试消息驱动bean客户端

* @version

* @Description:

* @author <ahref="mailto:zhenhuayue@sina.com">Retacn</a>

* @since 2016-7-24

public class DrivenBeanClient {

/**

* @param args

public static void main(String[] args) {

// 点对点模式消息测试

// p2p();

// 发布定阅模式消息测试

topic();

}

public static void p2p() {

final String beanName = "JmsMessP2PProducerBean";

final String viewClassName = JmsMessP2PProducer.class.getName();

try {

JmsMessP2PProducer p2pProducer =(JmsMessP2PProducer) lookupRemoteStatelessEjbBean(

beanName, viewClassName);

System.out.println(p2pProducer);

p2pProducer.send("这是一条点对点模式的的文本消息!");

} catch (NamingException e) {

e.printStackTrace();

}

public static void topic() {

final String beanName = "JmsMessTopicProducerBean";

final String viewClassName = JmsMessTopicProducer.class.getName();

try {

JmsMessTopicProducer topicProducer= (JmsMessTopicProducer) lookupRemoteStatelessEjbBean(

beanName,

viewClassName);

System.out.println(topicProducer);

topicProducer.send("这是一条发布定阅模式的文本消息!");

} catch (NamingException e) {

e.printStackTrace();

}

@SuppressWarnings("unchecked")

private static JmsMessProducer lookupRemoteStatelessEjbBean(StringbeanName,

String viewClassName) throws NamingException{

@SuppressWarnings("rawtypes")

final Hashtable jndiProperties = new Hashtable();

jndiProperties.put(Context.URL_PKG_PREFIXES,"org.jboss.ejb.client.naming");

final Context context = new InitialContext(jndiProperties);

// 如果服务器布署的是.ear包,则appname为.ear之后的包名,如果的war或是jar包,appName可以为空

final String appName = "";

final String moduleName = "HelloWorldEJB";

final String distinctName = "";

final String namespace = "ejb:" + appName + "/" +moduleName + "/" + distinctName + "/"

+ beanName + "!" +viewClassName;

System.out.println("namespace:" + namespace);

return (JmsMessProducer) context.lookup(namespace);

}

查看jboss控制台,输出结果如下:

点对点模式

10:57:13,264 INFO [stdout] (EJB default - 1)JmsMessP2PProducerBean: 发送的消息为: 这是一条点对点模式的的文本消息!

10:57:13,278 INFO [stdout] (Thread-5(HornetQ-client-global-threads-17012033)) JmsMessP2PConsumer: 接收到的消息为: 这是一条点对点模式的的文本消息

发布定阅模式

11:03:00,522 INFO [stdout] (EJB default - 2)JmsMessTopicProducerBean 发送的消息为: 这是一条发布定阅模式的文本消息!

11:03:00,529 INFO [stdout] (Thread-8(HornetQ-client-global-threads-17012033)) JmsMessTopicConsumerOne 接收到的消息为: 这是一条发布定阅模式的

文本消息!

11:03:00,539 INFO [stdout] (Thread-9(HornetQ-client-global-threads-17012033)) JmsMessTopicConsumerTwo 接收到的消息为: 这是一条发布定阅模式的

文本消息!