JavaEE(11) - 消息驱动EJB

时间：2023-11-16 23:58:20

1. MDB作为异步消费者的本质

2. MDB的运行机制

3. 使用@MessageDriven修饰MDB(需要messageListenerInterface)

4. 实现MessageListener

5. MDB的生命周期

6. 在MDB中使用依赖注入

7. MDB中的事务管理和异常管理

------------------------------------------------

1. MDB作为异步消费者的本质

MDB存在于EJB容器之中，可以利用EJB提供的事务、安全和并发性等系统服务。MDB并不直接与客户端交互，只是一个JMS消息的异步消费者。

EAO: Entiry Access Object

2. MDB的运行机制

MDB是由无状态Session Bean变化而来的，因此在用法上与其相似，不会保存客户端的调用状态，可被多个客户端共享。客户端需要以同步方式来调用无状态Session Bean，MDB不允许直接调用，它只是一个消息监听者。

消息生产者(发送消息)-->JMS消息目的(触发onMessage()方法)-->MDB(调用业务方法)-->Session Bean

3. 使用@MessageDriven修饰MDB

#1. EJB开发(Net Beans创建EJB Module, 项目名称：SimpleMDB)

SimpleMDB.java

package org.crazyit.jms;

import javax.ejb.*;

import javax.jms.*;

@MessageDriven(activationConfig

        = {

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

            @ActivationConfigProperty(propertyName = "acknowledgeMode", propertyValue = "Auto-acknowledge"),

            @ActivationConfigProperty(propertyName = "destination", propertyValue = "MessageQueue")

        } ,

        messageListenerInterface = javax.jms.MessageListener.class,

        mappedName = "MessageQueue"

)

public class SimpleMDB {

    public void onMessage(Message msg) {

        try {

            if (msg instanceof TextMessage) {

                TextMessage txt = (TextMessage) msg;

                String content = txt.getText();

                System.out.println("JMS信息中信息为：" + content);

            }

        }

        catch (Exception ex) {

            ex.printStackTrace();

        }

    }

}

#2. 消息发送(Net Beans创建Java Project, 项目名称：SendMsg)

SendMsg.java

package lee;

import javax.jms.*;

import javax.naming.*;

import java.util.Properties;

public class SendMsg {

    public void sendMessage() throws NamingException, JMSException {

        //定义WebLogic默认连接工厂的JNDI

        final String CONNECTION_FACTORY_JNDI = "weblogic.jms.ConnectionFactory";

        //获取JNDI服务所需的Context

        Context ctx = getInitialContext();

        //通过JNDI查找获取连接工厂

        ConnectionFactory connFactory = (ConnectionFactory) ctx.lookup(CONNECTION_FACTORY_JNDI);

        //通过JNDI查找获取消息目的

        Destination dest = (Destination) ctx.lookup("MessageQueue");

        //连接工厂创建连接

        Connection conn = connFactory.createConnection();

        //JMS连接创建JMS会话

        Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);

        //JMS会话创建消息生产者

        MessageProducer sender = session.createProducer(dest);

        //设置消息生产者生产出来的消息的传递模式、有效时间。

        sender.setDeliveryMode(DeliveryMode.PERSISTENT);

        sender.setTimeToLive(20000);

        //通过JMS会话创建一个文本消息

        TextMessage msg = session.createTextMessage();

    　　//msg.setStringProperty("ConType","txt");

        //设置消息内容

        msg.setText("Hello");

        //发送消息

        sender.send(msg);

        msg.setText("Welcome to JMS");

        //再次发送消息

        sender.send(msg);

        //关闭资源

        session.close();

        conn.close();

    }

    //工具方法，用来获取命名服务的Context对象

    private Context getInitialContext() {

        // 参见(4)

    }

    public static void main(String[] args) throws Exception {

        SendMsg sender = new SendMsg();

        sender.sendMessage();

    }

}

SendMapMsg.java

package lee;

import javax.jms.*;

import javax.naming.*;

import java.util.Properties;

public class SendMapMsg {

    public void sendMessage() throws NamingException, JMSException {

        //定义WebLogic默认连接工厂的JNDI

        final String CONNECTION_FACTORY_JNDI = "weblogic.jms.ConnectionFactory";

        //获取JNDI服务所需的Context

        Context ctx = getInitialContext();

        //通过JNDI查找获取连接工厂

        ConnectionFactory connFactory = (ConnectionFactory) ctx.lookup(CONNECTION_FACTORY_JNDI);

        //通过JNDI查找获取消息目的

        Destination dest = (Destination) ctx.lookup("MessageQueue");

        //连接工厂创建连接

        Connection conn = connFactory.createConnection();

        //JMS连接创建JMS会话

        Session session = conn.createSession(false/*不是事务性会话*/, Session.AUTO_ACKNOWLEDGE);

        //JMS会话创建消息生产者

        MessageProducer sender = session.createProducer(dest);

        //设置消息生产者生产出来的消息的传递模式、有效时间。

        sender.setDeliveryMode(DeliveryMode.PERSISTENT);

        sender.setTimeToLive(20000);

        //通过JMS会话创建一个文本消息

        MapMessage msg = session.createMapMessage();

        //设置消息内容

        msg.setString("name", "孙悟空");

        msg.setString("gender", "男");

        msg.setInt("age", 500);

        //发送消息

        sender.send(msg);

        //关闭资源

        session.close();

        conn.close();

    }

    //工具方法，用来获取命名服务的Context对象

    private Context getInitialContext() {

        // 参见(4)

    }

    public static void main(String[] args) throws Exception {

        SendMapMsg sender = new SendMapMsg();

        sender.sendMessage();

    }

}

4. 实现MessageListener

#1. EJB开发(Net Beans创建EJB Module, 项目名称：SimpleMDBListener)

SimpleMDB.java

package org.crazyit.jms;

import javax.ejb.ActivationConfigProperty;

import javax.ejb.MessageDriven;

import javax.jms.Message;

import javax.jms.MessageListener;

import javax.jms.TextMessage;

@MessageDriven(activationConfig ={

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

            @ActivationConfigProperty(propertyName = "acknowledgeMode", propertyValue = "Auto-acknowledge"),

            @ActivationConfigProperty(propertyName = "destination", propertyValue = "MessageQueue")

        },
　　　　 mappedName = "MessageQueue"

)

public class SimpleMDB implements MessageListener {

    @Override

    public void onMessage(Message msg) {

        try {

            if (msg instanceof TextMessage) {

                TextMessage txt = (TextMessage) msg;

                String content = txt.getText();

                System.out.println("JMS信息中信息为：" + content);

            }

        }

        catch (Exception ex) {

            ex.printStackTrace();

        }

    }

}

5. MDB的生命周期

#1. EJB开发(Net Beans创建EJB Module, 项目名称：Lifecycle)

LifecycleMDB.java

package org.crazyit.jms;

import javax.ejb.*;

import javax.jms.*;

import javax.annotation.*;

@MessageDriven(activationConfig

        = {

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

            @ActivationConfigProperty(propertyName = "acknowledgeMode", propertyValue = "Auto-acknowledge"),

            @ActivationConfigProperty(propertyName = "destination", propertyValue = "MessageQueue")

        },

        mappedName = "MessageQueue"

)

public class LifecycleMDB implements MessageListener {

    public void onMessage(Message msg) {

        try {

            if (msg instanceof TextMessage) {

                TextMessage txt = (TextMessage) msg;

                String content = txt.getText();

                System.out.println("JMS信息中信息为：" + content);

            }

        }

        catch (Exception ex) {

            ex.printStackTrace();

        }

    }

    @PostConstruct

    public void myInit() {

        System.out.println("--初始化方法--");

    }

    @PreDestroy

    public void myDestroy() {

        System.out.println("--销毁之前的方法--");

    }

}

6. 在MDB中使用依赖注入

#1. EJB开发(Net Beans创建EJB Module, 项目名称：Injection)

Injection.java

package org.crazyit.jms;

import javax.ejb.*;

import javax.jms.*;

import org.crazyit.service.*;

@MessageDriven(activationConfig

        = {

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

            @ActivationConfigProperty(propertyName = "acknowledgeMode", propertyValue = "Auto-acknowledge"),

            @ActivationConfigProperty(propertyName = "destination", propertyValue = "MessageQueue")

        },

        mappedName = "MessageQueue"

)

public class Injection implements MessageListener {

    @EJB(name = "StudentBean")

    private Student student;

    public void onMessage(Message msg) {

        try {

            if (msg instanceof MapMessage) {

                MapMessage map = (MapMessage) msg;

                String name = map.getString("name");

                String gender = map.getString("gender");

                int age = map.getInt("age");

                //调用Session Bean的方法添加学生。

                student.add(name, gender, age);

            }

        }

        catch (Exception ex) {

            ex.printStackTrace();

        }

    }

}

Student.java

package org.crazyit.service;

import javax.ejb.*;

@Remote

public interface Student {

    void add(String name, String gender, int age) throws Exception;

}

StudentBean.java

package org.crazyit.service;

import java.sql.*;

import javax.sql.*;

import javax.ejb.*;

import javax.annotation.*;

@Stateless(name = "StudentBean")

public class StudentBean implements Student {

    //采用依赖注入获取数据源

    @Resource(name = "javaee")

    private DataSource ds;

    public void add(String name, String gender, int age) throws Exception {

        Connection conn = null;

        PreparedStatement pstmt = null;

        try {

            //通过数据源获取数据库连接

            conn = ds.getConnection();

            //使用PreparedStatement执行SQL语句

            pstmt = conn.prepareStatement("insert into student values(null , ? , ? , ?)");

            pstmt.setString(1, name);

            pstmt.setString(2, gender);

            pstmt.setInt(3, age);

            pstmt.executeUpdate();

        }

        finally {

            pstmt.close();

            conn.close();

        }

    }

}

7. MDB中的事务管理和异常管理

MDB同样支持CMT和BMT，使用CMT，采用如下两种Annotation来修饰MDB的Bean实现类或方法：

@TransactionManagement：配置事务管理类型

@TransactionAttribute：配置事务管理属性

如果希望CMT事务管理机制遇到自定义异常时也能回滚事务，有两种处理方式：

#1. 定义该自定义异常类时使用@ApplicationException(rollback=true)

#2. 程序中显式捕获该异常，然后调用ctx.setRollback(true);代码来控制事务回滚。

相关文章