EJB3.0 MDB的实现



EJB3.0 MDB的实现

2009年02月05日 星期四 11:29

首先消息驱动bean是一个消息的接收者,它要实现MessageListioner接口,并且实现里面onMessage方法,作为ejb3来说需要annotation来说明这个bean是mdb
@MessageDriven(activationConfig = {
        @ActivationConfigProperty(propertyName = "destinationType", propertyValue = "javax.jms.Queue"),
        @ActivationConfigProperty(propertyName = "destination", propertyValue = "queue/myDestination")})

第一个属性标识,这个mdb是个基于队列的
第二个属性标识,这个mdb的目标位置,也就是JNDI
然后就可以部署这个MDB了,下面是我的源程序
t javax.ejb.ActivationConfigProperty;
import javax.ejb.MessageDriven;
import javax.jms.BytesMessage;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;
import javax.jms.StreamMessage;
import javax.jms.TextMessage;

@MessageDriven(activationConfig = {
        @ActivationConfigProperty(propertyName = "destinationType", propertyValue = "javax.jms.Queue"),
        @ActivationConfigProperty(propertyName = "destination", propertyValue = "queue/myDestination")})
publicclass Messager implements MessageListener
{

    publicvoid onMessage(Message msg)
    {
        try
        {
            if (msg instanceof TextMessage)
            {
                TextMessage tmsg = (TextMessage) msg;
                String content = tmsg.getText();
                this.print(content);
            } elseif (msg instanceof ObjectMessage)
            {
                ObjectMessage omsg = (ObjectMessage) msg;
                Man man = (Man) omsg.getObject();
                String content = man.getName() + " 家住" + man.getAddress();
                this.print(content);
            } elseif (msg instanceof MapMessage)
            {
                MapMessage map = (MapMessage) msg;
                String content = map.getString("no1");
                this.print(content);
            } elseif (msg instanceof BytesMessage)
            {
                BytesMessage bmsg = (BytesMessage) msg;
                ByteArrayOutputStream byteStream = new ByteArrayOutputStream();
                byte[] buffer = newbyte[256];
                int length = 0;
                while ((length = bmsg.readBytes(buffer)) != -1)
                {
                    byteStream.write(buffer, 0, length);
                }
                String content = new String(byteStream.toByteArray());
                byteStream.close();
                this.print(content);
            } elseif (msg instanceof StreamMessage)
            {
                StreamMessage smsg = (StreamMessage) msg;
                String content = smsg.readString();
                this.print(content);
            }
        } catch (Exception e)
        {
            e.printStackTrace();
        }

    }

    privatevoid print(String content)
    {
        System.out.println(content);
    }

}
这是里面用到的一个pojo
import java.io.Serializable;

publicclass Man implements Serializable
{
    privatestaticfinallongserialVersionUID = -1789733418716736359L;

    private String name;//姓名

    private String address;//地址

    public Man(String name, String address)
    {
        this.name = name;
        this.address = address;
    }

    public String getName()
    {
        returnname;
    }

    publicvoid setName(String name)
    {
        this.name = name;
    }

    public String getAddress()
    {
        returnaddress;
    }

    publicvoid setAddress(String address)
    {
        this.address = address;
    }
}
打成jar包并且部署
控制台出错,大概的意思是说我在annotation中标识的destination的JNDI没有找到.
因此需要写一个配置文件来标识我的mdb的jndi
起名为mymdb-service.xml并放于deploy目录下
文件内容<?xml version="1.0" encoding="UTF-8"?>
<server>
   <mbean code="org.jboss.mq.server.jmx.Queue"
      name="jboss.mq.destination:service=Queue,name=myDestination">
      <attribute name="JNDIName">queue/myDestination</attribute>
      <depends optional-attribute-name="DestinationManager">jboss.mq:service=DestinationManager</depends>
   </mbean>
</server>

然后再启动,控制台正常.

书写客户端调用程序
package test.jms;

import java.util.Properties;

import javax.jms.BytesMessage;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.QueueSession;
import javax.jms.StreamMessage;
import javax.jms.TextMessage;
import javax.naming.InitialContext;

import com.foshanshop.ejb3.jms.Man;

publicclass QueueSender
{

    publicstaticvoid main(String[] args)
    {
        QueueConnection conn = null;
        QueueSession session = null;
        try
        {
            Properties props = new Properties();
            props.setProperty("java.naming.factory.initial",
                    "org.jnp.interfaces.NamingContextFactory");
            props.setProperty("java.naming.provider.url", "localhost:1099");
            props.setProperty("java.naming.factory.url.pkgs",
                    "org.jboss.naming:org.jnp.interfaces");
            InitialContext ctx = new InitialContext(props);
            QueueConnectionFactory factory = (QueueConnectionFactory) ctx
                    .lookup("ConnectionFactory");
            conn = factory.createQueueConnection();
            session = conn.createQueueSession(false,
                    QueueSession.AUTO_ACKNOWLEDGE);
            Destination destination = (Queue) ctx.lookup("queue/myDestination");
            MessageProducer producer = session.createProducer(destination);
            //        发送文本
            TextMessage msg = session
                    .createTextMessage("这是我的第一个消息驱动Bean");
            producer.send(msg);
            //        发送Ojbect(对像必须实现序列化,否则等着出错吧)
            producer
                    .send(session.createObjectMessage(new Man("美女", "北京")));
            //        发送MapMessage
            MapMessage mapmsg = session.createMapMessage();
            mapmsg.setObject("no1", "北京");
            producer.send(mapmsg);
            //        发送BytesMessage
            BytesMessage bmsg = session.createBytesMessage();
            bmsg.writeBytes("我是一个兵,来自老百姓".getBytes());
            producer.send(bmsg);//发送StreamMessage
            StreamMessage smsg = session.createStreamMessage();
            smsg.writeString("我就爱流读写");
            producer.send(smsg);
        } catch (Exception e)
        {
            System.out.println(e.getMessage());
        } finally
        {
            try
            {
                session.close();
                conn.close();
            } catch (JMSException e)
            {
                e.printStackTrace();
            }
        }
    }
}