EJB_消息驱动发展bean

时间:2021-08-08 02:50:26

消息驱动发展bean

Java信息服务(Java MessageService)

Java 信息服务(Java Message Service,简称 JMS)是用于訪问企业消息系统的开发商中立的API。

企业消息系统能够协助应用软件通过网络进行消息交互。

JMS的编程过程非常easy,概括为:应用程序A发送一条消息到消息server的某个目地(Destination),然后消息server把消息转发给应用程序B。

由于应用程序A和应用程序B没有直接的代码关连,所以两者实现了解偶。

JMS中的消息

消息传递系统的中心就是消息。一条 Message 由三个部分组成:

头(header),属性(property)和主体(body)。

消息有以下几种类型。他们都是派生自 Message 接口。

StreamMessage:一种主体中包括 Java 基元值流的消息。其填充和读取均按顺序进行。

MapMessage:一种主体中包括一组名-值对的消息。

未定义条目顺序。

TextMessage:一种主体中包括 Java 字符串的消息(比如,XML 消息)。

ObjectMessage:一种主体中包括序列化 Java 对象的消息。

BytesMessage:一种主体中包括连续字节流的消息。

消息的传递模型

JMS 支持两种消息传递模型:点对点(point-to-point。简称 PTP)和公布/订阅(publish/subscribe,简称 pub/sub)。

这两种消息传递模型很相似。但有下面差别:

PTP 消息传递模型规定了一条消息仅仅能传递给一个接收方。 採用javax.jms.Queue 表示。

Pub/sub 消息传递模型同意一条消息传递给多个接收方。

採用javax.jms.Topic表示

这两种模型都通过扩展公用基类来实现。比如:javax.jms.Queue 和javax.jms.Topic 都扩展自javax.jms.Destination 类。

点对点PTP 消息传递模型

EJB_消息驱动发展bean

谁先监听,谁就先收到。

公布/订阅消息传递模型

 EJB_消息驱动发展bean

watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvSmVyb21lX3M=/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/Center" alt="">

配置目标地址

開始JMS编程前。我们须要先配置消息到达的目标地址(Destination)。由于仅仅有目标地址存在了,我们才干发送消息到这个地址。由于每一个应用server关于目标地址的配置方式都有所不同,以下以jboss为例,配置一个queue类型的目标地址。

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

<server>

<mbeancode="org.jboss.mq.server.jmx.Queue"

name="jboss.mq.destination:service=Queue,name=foshanshop">

 <attributename="JNDIName">queue/foshanshop</attribute>  

<dependsoptional-attribute-name="DestinationManager">jboss.mq:service=DestinationManager</depends>

</mbean>

</server>

Jboss使用一个XML文件配置队列地址。文件的取名格式应遵守*-service.xml

<attribute name="JNDIName">属性指定了该目标地址的全局JNDI名称。假设你不指定JNDIName属性,jboss会为你生成一个默认的全局JNDI,其名称由“queue”+“/”+目标地址名称组成。

另外在不论什么队列或主题被部署之前,应用server必须先部署Destination Manager Mbean。所以我们通过<depends>节点声明这一依赖。

加粗自己定义设置,黑色不用更改。

步骤:

在桌面新建一个hqu-service.xml复制上面xml代码改动加粗部分:

<?

xml
version="1.0"encoding="UTF-8"?>

<server>

<mbean
code="org.jboss.mq.server.jmx.Queue"

name="jboss.mq.destination:service=Queue,name=hquQueue">

<attribute
name="JNDIName">queue/hquQueue</attribute>

<depends
optional-attribute-name="DestinationManager">

jboss.mq:service=DestinationManager</depends>

</mbean>

</server>

公布:复制到Jboss的server\default\deploy下就公布了。能够在控制台看到公布信息。

能够在Jboss管理页面看:http://localhost:8080/jmx-console/

jboss.mq.destination 下的name=hquQueue,service=Queue

目标地址建立完毕后。我们就能够进行消息的发送。

在Java类中发送消息

一般发送消息有下面步骤:

(1) 得到一个JNDI初始化上下文(Context)

InitialContextctx = new InitialContext();

(2) 依据上下文查找一个连接工厂QueueConnectionFactory 。

该连接工厂是由JMS提供的,不需我们自己创建。每一个厂商都为它绑定了一个全局JNDI,我们通过它的全局JNDI便可获取它;

QueueConnectionFactoryfactory = (QueueConnectionFactory)

ctx.lookup("QueueConnectionFactory");

(3) 从连接工厂得到一个连接QueueConnection

conn= factory.createQueueConnection();

(4) 通过连接来建立一个会话(Session);

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

这句代码意思是:建立不须要事务的而且能自己主动确认消息已接收的会话。

(5) 查找目标地址:

样例相应代码:Destinationdestination = (Destination ) ctx.lookup("queue/hquQueue");

(6) 依据会话以及目标地址来建立消息生产者MessageProducer(QueueSender和TopicPublisher都扩展自MessageProducer接口)

样例相应代码:

MessageProducer producer =session.createProducer(destination);

TextMessage msg =session.createTextMessage("您好,这是我的第一个消息驱动Bean");

producer.send(msg);

步骤:

新建一个消息驱动bean项目,新建JavaProject:MessageDrivenBean。加入EJB需的jar文件, 新建一个java类:QueueSender在cn.hqu.app下,用于发送queue。

package cn.hqu.app;

import javax.jms.Destination;
import javax.jms.MessageProducer;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.QueueSession;
import javax.naming.InitialContext; public class QueueSender {
public static void main(String[] args) {
try {
//1.首先初始化一个JNDI上下文对象
InitialContext ctx = new InitialContext();
//2.依据上下文查找一个连接工厂 QueueConnectionFactory
QueueConnectionFactory factory = (QueueConnectionFactory) ctx.lookup("QueueConnectionFactory");
//3.通过连接工厂能够创建一个queue连接QueueConnection。
QueueConnection conn = factory.createQueueConnection();
//4.通过连接创建一个到目的地址的会话
QueueSession session = conn.createQueueSession(false, QueueSession.AUTO_ACKNOWLEDGE);
//5.查找目标地址
Destination destination = (Destination) ctx.lookup("queue/hquQueue");
//6.得到消息的发送者
MessageProducer producer = session.createProducer(destination);
//通过这个发送者我们就能够发送消息了:
producer.send(session.createTextMessage("你好,苏志达"));
session.close();
conn.close();
} catch (Exception e) {
e.printStackTrace();
}
}
}

在方法内部。编写发送代码。

1.      首先初始化一个JNDI上下文对象:

拷贝上一个项目的jdni.properties到src下,在类QueueSender的mian里面

try {

InitialContextctx = newInitialContext();

}catch(NamingException e) {

e.printStackTrace();

}

2.依据上下文查找一个连接工厂 QueueConnectionFactory

查找到queue类型的连接工厂,

QueueConnectionFactory factory =(QueueConnectionFactory) ctx.lookup("QueueConnectionFactory");

3.通过连接工厂能够创建一个queue连接QueueConnection。

QueueConnectionconn = factory.createQueueConnection();

4.通过连接创建一个到目的地址的会话:第一个參数指定是否须要事务。第二个參数指定消息确定模式。

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

5.查找目标地址:

Destination destination= (Destination) ctx.lookup("queue/hquQueue");

6.得到消息的发送者

MessageProducerproducer = session.createProducer(destination);

通过这个发送者我们就能够发送消息了:

producer.send(session.createTextMessage("你好。苏志达"));

调用程序往目标地址发送一条消息。执行main。消息发送完。接下来编写消息接收者。

採用消息驱动Bean接收消息

能够採用Java类的形式进行接收, 这里採用消息驱动Bean(Message Driven Bean)接收消息,由于消息驱动Bean(MDB)是设计用来专门处理基于消息请求的组件。它和无状态Session Bean一样也使用了实例池技术。容器能够使用一定数量的Bean实例并发处理成百上千个JMS消息

正由于MDB具有处理大量并发消息的能力。所以非常适合应用在一些消息网关产品。假设一个业务运行的时间非常长,而运行结果无需实时向用户反馈时。也非常适合使用MDB。如订单成功后给用户发送一封电子邮件或发送一条短信等。

一个MDB通常要实现MessageListener接口,该接口定义了onMessage()方法。Bean通过它来处理收到的JMS消息。

package javax.jms;

public interface MessageListener {

public void onMessage(Message message);

}

当容器检測到Bean守候的目标地址有消息到达时,容器调用onMessage()方法,将消息作为參数传入MDB。MDB在onMessage()中决定怎样处理该消息。你能够使用凝视指定MDB监听哪一个目标地址(Destination)。当MDB部署时。容器将读取当中的配置信息。

步骤:

新建一个消息驱动Bean:MessageDrivenBean在cn.hqu.message下实现MessageListener。

当容器检測到这个JavaBean所监听的目标地址有消息到达的时候,会帮我们获得这个消息传到onMessage这种方法。

通过注解能够指定消息驱动bean它监听目标地址的类型是什么类型,和目标地址的JNDI名称。

@MessageDriven(activationConfig=

{

@ActivationConfigProperty(propertyName="destinationType",

propertyValue="javax.jms.Queue"),

@ActivationConfigProperty(propertyName="destination",

propertyValue="queue/hquQueu"),

@ActivationConfigProperty(propertyName="acknowledgeMode",

propertyValue="Auto-acknowledge")

})

public
class
MessageDrivenBean implements MessageListener{

从queue类型的目标地址去寻找JNDI为queue/hquQueu所绑定的目的地址。

消息接收到的时候消息的确认模式为自己主动确认。acknowledgeMode是默认的所以这个能够去掉。

在onMessage方法里面处理接收到的消息

public
void
onMessage(Message msg) {

TextMessage message = (TextMessage) msg;

try {

System.out.println(message.getText());

} catch (JMSException e) {

e.printStackTrace();

}

}

进行打包,部署。

拷贝上一个项目的ant,改动配置文件build.xml的name为

MessageDrivenBean

运行deploy,把消息驱动bean公布到Jboss中。在控制台打印

16:07:57,999 INFO  [JmxKernelAbstraction] installing MBean:jboss.j2ee:jar=Messa

geDrivenBean.jar,name=MessageDrivenBean,service=EJB3with dependencies:

16:07:58,015 INFO  [EJBContainer] STARTED EJB:cn.hqu.message.MessageDrivenBean

ejbName: MessageDrivenBean

16:07:58,039 INFO  [EJB3Deployer] Deployed:file:/F:/Java/jboss-4.2.2.GA/server/

default/deploy/MessageDrivenBean.jar

16:07:58,060 INFO  [STDOUT] 你好,苏志达

接收到了消息。

Jms同意发送者和接收方能够不同一时候在线。两者之间没有代码的关联,实现了非常好的解耦。

代码:http://pan.baidu.com/s/1bn2DgPT

关于queue类型的发送就完了,接下来关于

Topic的发送和接收。

第一步配置目标地址。类型不是queue类型,是Topic类型,在配置文件hqu-service.xml加一个Topic配置

<mbean code="org.jboss.mq.server.jmx.Topic"

name="jboss.mq.destination:service=Topic,name=hquTopic">

<attribute name="JNDIName">topic/hquTopic</attribute>

<depends optional-attribute-name="DestinationManager">

jboss.mq:service=DestinationManager</depends>

</mbean>

配置好,将它公布到Jboss中(复制进去)。

新建类TopicSender在cn.hqu.app下,用于公布topic 类型的消息。

JNDI名称能够在公布后的控制台看到。

package cn.hqu.app;

import javax.jms.Destination;
import javax.jms.MessageProducer;
import javax.jms.TopicConnection;
import javax.jms.TopicConnectionFactory;
import javax.jms.TopicSession;
import javax.naming.InitialContext; public class TopicSender { /**
* @param args
*/
public static void main(String[] args) {
try {
//1.首先初始化一个JNDI上下文对象
InitialContext ctx = new InitialContext();
//2.依据上下文查找一个连接工厂 QueueConnectionFactory
TopicConnectionFactory factory = (TopicConnectionFactory) ctx.lookup("TopicConnectionFactory");
//3.通过连接工厂能够创建一个queue连接QueueConnection。
TopicConnection conn = factory.createTopicConnection();
//4.通过连接创建一个到目的地址的会话
TopicSession session = conn.createTopicSession(false, TopicSession.AUTO_ACKNOWLEDGE);
//5.查找目标地址
Destination destination = (Destination) ctx.lookup("topic/hquTopic");
//6.得到消息的发送者
MessageProducer producer = session.createProducer(destination);
//通过这个发送者我们就能够发送消息了:
producer.send(session.createTextMessage("你好,苏志达"));
session.close();
conn.close();
} catch (Exception e) {
e.printStackTrace();
}
} }

执行main。

开发消息的接收方。由于会被多方接收,所以建立两个消息驱动Bean

新建ReceiveBean实现MessageListener在cn.hqu.message下。

package cn.hqu.message;

import javax.ejb.ActivationConfigProperty;
import javax.ejb.MessageDriven;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
@MessageDriven(activationConfig =
{
@ActivationConfigProperty(propertyName="destinationType",
propertyValue="javax.jms.Topic"),
@ActivationConfigProperty(propertyName="destination",
propertyValue="topic/hquTopic")
})
public class ReceiveBean implements MessageListener { @Override
public void onMessage(Message msg) {
TextMessage message = (TextMessage) msg;
try {
System.out.println(this.getClass()+message.getText());
} catch (JMSException e) {
e.printStackTrace();
}
} }

另外一个消息驱动Bean:ReceiveOtherBean

package cn.hqu.message;

import javax.ejb.ActivationConfigProperty;
import javax.ejb.MessageDriven;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
@MessageDriven(activationConfig =
{
@ActivationConfigProperty(propertyName="destinationType",
propertyValue="javax.jms.Topic"),
@ActivationConfigProperty(propertyName="destination",
propertyValue="topic/hquTopic")
})
public class ReceiveOtherBean implements MessageListener { @Override
public void onMessage(Message msg) {
TextMessage message = (TextMessage) msg;
try {
System.out.println(this.getClass()+message.getText());
} catch (JMSException e) {
e.printStackTrace();
}
} }

公布执行Ant deploy

公布完毕并没有看到刚才发送的topic消息,由于对于topic类型的消息,假设当时接收方没有监听着这个topic类型的消息。他就获取不到这个消息。虽然我们如今把它部署到Jboss,由于当时它并没有监听topic目标地址,所以他也是无法得到这个消息的。

如今我们再运行一下topic的发送,就能够在控制台看到多个接收方接受到发送的消息了。

代码:http://pan.baidu.com/s/1nthubNF

版权声明:本文博客原创文章,博客,未经同意,不得转载。