JMS API 2.0 连接何时关闭？

Question

我正在使用 JMS API 简化版 2.0，但遇到了问题。我通过发送消息

messageProducer = context.createProducer();

messageProducer.send(destination,message);

在此之前，我通过此代码创建消息

ObjectMessage msg = context.createObjectMessage();

TemporaryQueue tempQueue = context.createTemporaryQueue();

messageConsumer = context.createConsumer(tempQueue);

msg.setObject(foo);

msg.setJMSReplyTo(tempQueue); 

msg.setJMSCorrelationID(getRandomString()); 

消息已正确接收到 MDB。

在 mdb 中我创建了一个响应字符串

TextMessage msg = context.createTextMessage();

msg.setText("message response");
msg.setJMSCorrelationID(); here the id correlation of the object received

然后创建要发送的消息

JMSProducer messageProducer = context.createProducer();

messageProducer.send(ReceiverTemporaryDestination, msg);

不幸的是程序在JMS端冻结到以下函数

Message msg = messageConsumer.receive();

我明白为什么它不能正常工作。当我使用临时队列时，似乎连接在发送响应消息之前就关闭了。我该如何解决这个问题？是否存在不关闭连接的方法？

JSM Class

@Stateless
public class BookingManagerBean implements BookingManagerBeanInterface
{
    @EJB
    private UtilBeanInterface utilBean;
    @EJB
    private ReservationManagerBeanInterface reservationBean;

    @Inject
    @JMSConnectionFactory("jms/reservationProcessorQueueFactory")
    private JMSContext context;

    @Resource(mappedName = "jms/bookingProcessorQueueReceiver")
    private Queue bookingProcessorQueueReceiver;

    private JMSProducer messageProducer = null;
    private JMSConsumer messageConsumer = null;

    private String id_reservation = null;

    private Map<String,Object> map;

    @Override
    public String purchase(Route r, UserCredential u, int passengers, int  luggages, double price, int points)
    {                         
        Reservation reserv = reservationBean.addReservation(r, u, passengers, luggages, (float) price);

        try
        {
            sendMessageToBookingProcessorBean(reserv, points); // invia il messaggio contenente la prenotazione e il numero di punti utilizzati                             

            receiveMessageFromBookingProcessorBean();
        }
        catch(JMSException jmse)
        {
            System.err.println("An error occured " + jmse.toString());
            return (id_reservation);
        }

        return id_reservation;
    }

    @TransactionAttribute(TransactionAttributeType.NOT_SUPPORTED)
    private void sendMessageToBookingProcessorBean(Object messageData, int points) throws JMSException
    {                                  
        messageProducer = context.createProducer();

        messageProducer.send(bookingProcessorQueueReceiver,createMessageForBookingProcessorBean(messageData, points));   
    }

    private Message createMessageForBookingProcessorBean(Object messageData, int points) throws JMSException
    {              
        ObjectMessage msg = context.createObjectMessage();

        TemporaryQueue tempQueue = context.createTemporaryQueue();

        messageConsumer = context.createConsumer(tempQueue);

        map = new HashMap<>();

        map.put("reservation", (Reservation) messageData);
        map.put("used_points", (Integer) points);

        msg.setObject((Serializable) map);

        msg.setJMSReplyTo(tempQueue); //setta  la coda temporanea dove arriverà la risposta
        msg.setJMSCorrelationID(utilBean.getRandomString()); //setta l'id 

        return msg;
    }

    @TransactionAttribute(TransactionAttributeType.NOT_SUPPORTED)
    private void receiveMessageFromBookingProcessorBean() throws JMSException
    {
        Message msg = messageConsumer.receive(10000); 

        if(msg instanceof TextMessage)
        {
            TextMessage response = (TextMessage)msg;

            this.id_reservation = response.getText();
        }
    }

}

MDB

@MessageDriven(mappedName = "jms/bookingProcessorQueueReceiver", 
               activationConfig = {@ActivationConfigProperty(propertyName = "acknowledgeMode", propertyValue = "Auto-acknowledge"),
                                   @ActivationConfigProperty(propertyName = "destinationType", propertyValue = "javax.jms.Queue")
})
public class BookingProcessorBean implements MessageListener
{   
    @EJB
    private ReservationManagerBeanInterface reservationManagerBean;
    @EJB
    private RouteManagerBeanInterface routeManagerBean;
    @EJB
    private SenderMailBeanInterface senderMailBean;
    @EJB
    private UserManagerBeanInterface userManagerBean;
    @EJB
    private GetterFidelityPointsBeanInterface getterFidelityPointsBean;

    @PersistenceContext
    private EntityManager em;

    @Inject
    @JMSConnectionFactory("jms/reservationProcessorQueueFactory")
    @JMSSessionMode(JMSContext.AUTO_ACKNOWLEDGE)
    private JMSContext context;

    Reservation reserv = null;

    private HashMap<String, Object> map = null;

    int points;

    @Override
    public void onMessage(Message message)
    {                                  
        try
        {       
            ObjectMessage om = (ObjectMessage) message;

            System.out.println(om.getJMSReplyTo());
            System.out.println(om.getJMSCorrelationID());


            map = (HashMap<String,Object>) om.getObject();

            reserv = (Reservation) map.get("reservation");
            points = (Integer) map.get("used_points");

            System.out.println("Processing reservation...");

            UserCredential user = reserv.getUsername(); 

            user.getReservationsList().add(reserv); 

            userManagerBean.updateFidelityPoints(user, points); 
            userManagerBean.addFidelityPoints(user, getterFidelityPointsBean.getFidelityPointsByRoute(reserv.getRoute())); 

            routeManagerBean.updateSeatsRoute(reserv); 

            reserv.setId(reservationManagerBean.generateIDReservation()); 

            storeInDb(reserv, user, reserv.getRoute()); 

            senderMailBean.sendEmail(reserv);

            TemporaryQueue tempQueue = (TemporaryQueue) om.getJMSReplyTo();

            sendIdReversationToBookingManagerBean(tempQueue, reserv.getId(), om.getJMSCorrelationID()); //prende la destinazione temporanea ed il messaggio per la risposta
        }
        catch(UnsupportedEncodingException uee)
        {
            System.err.println("An error occured during the sending of the confirmation mail to " + reserv.getUsername().getEmail());
        }
        catch(MessagingException me)
        {
            System.err.println("An error occured during the sending of the email to " + reserv.getUsername().getEmail());
        }
        catch(JMSException e)
        {
            System.err.println("An error occured during the processing of the booking " + reserv.getId());
        }
    }

    private void storeInDb(Reservation r, UserCredential u, Route route)
    {        
        em.persist(r);
        em.merge(u);
        em.merge(route);
    }

    private Message createMessageForBookingManagerBean(String id_reservation, String id_correlation) throws JMSException
    {
        TextMessage msg = context.createTextMessage();

        msg.setText(id_reservation);
        msg.setJMSCorrelationID(id_correlation);

        return msg;
    }

    @TransactionAttribute(TransactionAttributeType.NOT_SUPPORTED)
    private void sendIdReversationToBookingManagerBean(Destination ReceiverDestination, String id_reservation, String id_correlation) throws JMSException
    {
        JMSProducer messageProducer = context.createProducer();

        messageProducer.send(ReceiverDestination, createMessageForBookingManagerBean(id_reservation, id_correlation));
    }
}

Answer 1

如果您从 EJB 发送消息，请确保发送/接收方法具有 @TransactionAttribute(TransactionAttributeType.NOT_SUPPORTED)，因为正如 Robert 所说，它不能在一个事务中。

更新

在您的代码中更改以下内容： 在BookingManagerBean 中添加不支持的公共方法而不是私有方法：

@TransactionAttribute(TransactionAttributeType.NOT_SUPPORTED)
public String purchase(Route r, UserCredential u, int passengers, int  luggages,    double price, int points)

Answer 2

如果您在 Java EE 应用程序中工作，则无法在一个事务中发送和接收消息。 在事务提交之前，消息不会传递给接收者。 因此，如果您用于发送原始请求并从 MDB 接收响应的代码在一个事务中，则不应收到响应，因为 MDB 甚至不应收到请求，因为该事务之前尚未提交。