如果 bean 部署在同一个应用程序服务器中，一个线程用于更多 bean？

Question

我正在尝试使用 Java EE 实现（练习）主动复制算法。

我遇到了一个意想不到的问题。

我有这个功能（伪代码）：

public boolean doOperation() in the frontEnd EJB{
  sendJMSMessageToTopic();  //publish new command to the replica
  Thread.sleep(200);
  readFromQueue();          //read Ack from replica
  control if  write quorum is reached
  commit or abort
  return state (success or not)
}

问题是：

回调 onMessage() 已调用，但在 doOperation 的同一次调用中未收到响应。当我再次调用 doOperation 时，我读取了先前调用 doOperation 方法的响应。

我使用 glassfish，并将所有组件部署在同一个应用程序服务器中。我使用主题将请求发布到副本，并使用队列接收来自副本的响应。 FrontEnd 是 Singleton，Replicas 是 Message Drive Bean。

更新：我

我的副本代码

/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package EJBData;

import Service.ResponceMessage;
import Service.Command;
import Service.Product;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.annotation.Resource;
import javax.ejb.ActivationConfigProperty;
import javax.ejb.MessageDriven;
import javax.inject.Inject;
import javax.jms.JMSConnectionFactory;
import javax.jms.JMSContext;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;
import javax.jms.Queue;

/**
 *
 * @author melix
 */
@MessageDriven(activationConfig = {
    @ActivationConfigProperty(propertyName = "destinationType", propertyValue = "javax.jms.Topic"),
    @ActivationConfigProperty(propertyName = "destinationLookup", propertyValue = "TopicRepliche"),
    //@ActivationConfigProperty(propertyName = "subscriptionDurability", propertyValue = "durable"),
    @ActivationConfigProperty(propertyName = "clientId", propertyValue = "TopicRepliche"),
    @ActivationConfigProperty(propertyName = "subscriptionName", propertyValue = "TopicRepliche")
})
public class Replica1 implements MessageListener {
    @Resource(mappedName = "QueueFrontEnd")
    private Queue queueFrontEnd;
    @Inject
    @JMSConnectionFactory("java:comp/DefaultJMSConnectionFactory")
    private JMSContext context;


    private final int R_CONTAINS_KEY=0;
    private final int W_REMOVE_KEY=1;
    private final int W_PUT=2;
    private final int R_GET=3;
    private final int R_UPDATE_PRICE=4;
    private final int R_GENERATE_JSON_FROM_MAP=5;


    private final int COD_REPLICA=0;
    private final ConcurrentHashMap<Integer,Product> item_map=new ConcurrentHashMap<>();    

    private final ConcurrentHashMap<Long,Command> pending_command=new ConcurrentHashMap<>();
    private long last_cmd_executed=0;
    private long last_cmd_received=0;    


    private synchronized void setLastCmdExecuted(long last_cmd_executed){
     this.last_cmd_executed=last_cmd_executed;
    }

    private synchronized void setLastCmdReceived(long last_cmd_received){
     this.last_cmd_received=last_cmd_received;
    }


    public boolean containsKey(int key){
     return item_map.containsKey(key);
    }

    public boolean removeKey(int key){
     item_map.remove(key);
     return true;
    }

    public boolean put(Product b){
     item_map.put(Integer.parseInt(b.getId()),b);
     return true;
    }

    public Product get(int key){
     return item_map.get(key);
    }

    public void updatePrice(){
      for (Map.Entry pairs : item_map.entrySet()) {
            int key=(int)pairs.getKey();
            Product p=(Product)pairs.getValue();
            double price=p.getInitialPrice();
            item_map.remove(key);
            System.out.println(price);
            if(price>=5)
              p.setInitialPrice(price-5);
            else
              p.setInitialPrice(0);
            item_map.put(key, p);
        }
    }


    public String generateJSONFromMap(){
        String json="[";
        for (Map.Entry pairs : item_map.entrySet()) {
            Product p=(Product)pairs.getValue();
            json+=p.toString();
            json+=",";
        }

        if(json.endsWith(","))
            json = json.substring(0,json.length() - 1);
        json+="]";
        return json;
    }


    public Replica1() {
    }


    @Override
    public void onMessage(Message message) {
        Command c;
        if (message instanceof ObjectMessage) {    
            try {
                Object object = ((ObjectMessage) message).getObject();
                c=(Command)object;
                int command_type=c.getType();
                Object res=null;
                switch(command_type){
                    case R_CONTAINS_KEY:
                        res=containsKey((int)c.getInput());
                        break;
                    case W_REMOVE_KEY:
                        res=removeKey((int)c.getInput());
                        break;
                    case W_PUT:
                        res=put((Product)c.getInput());
                        break;
                    case R_GET:
                        res=get((int)c.getInput());
                        break;
                    case R_UPDATE_PRICE:
                        updatePrice();
                        break;
                    case R_GENERATE_JSON_FROM_MAP:
                        res=generateJSONFromMap();
                        break;
                }
                System.out.println("FROM REPLICA, ACK FOR"+c.getSqnCommand());
                this.setLastCmdReceived(c.getSqnCommand());
                sendAckToQueueFrontEnd(c.getSqnCommand(),res);
            } catch (JMSException ex) {
                Logger.getLogger(Replica1.class.getName()).log(Level.SEVERE, null, ex);
            }       
        }   
    }

    private void sendJMSMessageToQueueFrontEnd(String messageData) {
        context.createProducer().send(queueFrontEnd, messageData);
    }

    private void sendAckToQueueFrontEnd(long sqn_command,Object responce) {
        try {
            ResponceMessage ack=new ResponceMessage(true,false,sqn_command,COD_REPLICA);
            ack.setResponce(responce);
            ObjectMessage objectMessage=context.createObjectMessage();
            objectMessage.setObject(ack);
            context.createProducer().send(queueFrontEnd,objectMessage);
        } catch (JMSException ex) {
            Logger.getLogger(Replica1.class.getName()).log(Level.SEVERE, null, ex);
        }
    }   
}

前端代码： （doOperation 在我的代码中与 sendNewCommand() 映射）

/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package EJBData;

import EJBExecutor.ReaderLocal;
import Service.Command;
import Service.Product;
import Service.ResponceMessage;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.annotation.Resource;
import javax.ejb.EJB;
import javax.ejb.Stateless;
import javax.ejb.Timeout;
import javax.ejb.Timer;
import javax.ejb.TimerConfig;
import javax.ejb.TimerService;
import javax.inject.Inject;
import javax.jms.JMSConnectionFactory;
import javax.jms.JMSContext;
import javax.jms.JMSException;
import javax.jms.ObjectMessage;
import javax.jms.Topic;

/**
 *
 * @author melix
 */
@Stateless
public class AuctionFrontEnd implements AuctionFrontEndLocal{
    @EJB
    private FrontEndConsumer frontEndConsumer;

    @EJB
    private ReaderLocal reader;

    @Resource(mappedName = "TopicRepliche")
    private Topic topicRepliche;
    @Inject
    @JMSConnectionFactory("java:comp/DefaultJMSConnectionFactory")
    private JMSContext context;
    @Resource
    TimerService service;

    private boolean isConsumerInit=false;

    private final int R_CONTAINS_KEY=0;
    private final int W_REMOVE_KEY=1;
    private final int W_PUT=2;
    private final int R_GET=3;
    private final int R_UPDATE_PRICE=4;
    private final int R_GENERATE_JSON_FROM_MAP=5;


    private final ConcurrentHashMap<Integer,Product> item_map=new ConcurrentHashMap<>();
    private ArrayList<ConcurrentHashMap<Long,ResponceMessage>> listOfHashMap;   
    private Queue<Command> pending_command = new LinkedList<Command>();

    private final int WRITE_QUORUM=2;
    private final int READ_QUORUM=2;
    private final int NUM_REPLIC=3;

    private int last_key_inserted=0;
    private long num_command=0;

    public void initConsumer(){
      frontEndConsumer.init();
    }

    private synchronized void putCommandInQueue(Command c){
     pending_command.add(c);
    }

    private synchronized Command removeCommandFromQueue(Command c){
     if(!pending_command.isEmpty())
         return pending_command.remove();
     return null;
    }

    private synchronized void addAck(int cod_replica,long num_command,ResponceMessage m){
     listOfHashMap.get(cod_replica).put(num_command,m);
    }


    private synchronized void addResponce(int cod_replica,long num_command,ResponceMessage m){
     listOfHashMap.get(cod_replica+3).put(num_command,m);
    } 

    @Override
    public synchronized void addAckToList(int cod_replica,long num_command,ResponceMessage m){
     addAck(cod_replica,num_command,m);
    }

    @Override
    public synchronized void addResponceToList(int cod_replica,long num_command,ResponceMessage m){
     addAck(cod_replica,num_command,m);
    } 

    private synchronized long addNumCommand(){
     this.num_command++;
     return num_command;
    }

    @Override
    public void mexReceived(String message){
     System.out.println(message);   
    }

   @Timeout
    public void handleTimeout(Timer timer) {
        if(!pending_command.isEmpty()){
            Command c=pending_command.poll();
            for(int i=0;i<NUM_REPLIC*2;i++){
             if(listOfHashMap.get(i).containsKey(c.getSqnCommand())){
              ResponceMessage m=listOfHashMap.get(i).get(c.getSqnCommand());
              System.out.println("Ack per comando:"+m.getSqnCommand()+"from replica"+m.getCode_replica());
             }
            }
         }
        timer.cancel();
    }

    public void startTimer() {
        TimerConfig config = new TimerConfig();
        config.setPersistent(false);
        Timer timer = service.createSingleActionTimer(200, config);
    }




    private ResponceMessage[] sendNewCommand(boolean isWriteOperation,int type,Object input){
            int num_ack=0,i=0;
            Command c=new Command(isWriteOperation,addNumCommand(),type,input);
            ObjectMessage objectMessage=context.createObjectMessage();
            try {
                objectMessage.setObject(c);
            } catch (JMSException ex) {
                Logger.getLogger(AuctionFrontEnd.class.getName()).log(Level.SEVERE, null, ex);
            }
            if(!isConsumerInit){
             this.initConsumer();
             isConsumerInit=true;
            }   
            frontEndConsumer.cleanQueue();
            sendJMSMessageToTopicRepliche(objectMessage);            
            ResponceMessage[] m=new ResponceMessage[NUM_REPLIC];
            ResponceMessage tmp = null;
            do{
                tmp=frontEndConsumer.consume();                
                if(tmp!=null){
                 System.out.println("ACK CORRECT"+tmp.getSqnCommand()+";"+c.getSqnCommand());
                 if(tmp.getSqnCommand()==c.getSqnCommand()){
                  m[num_ack]=tmp; 
                  num_ack++;
                 }
                }
             }while(tmp!=null);
            System.out.println("sono alla fine!");
            if(isWriteOperation&&num_ack>=WRITE_QUORUM)
             return m;
            if(!isWriteOperation&&num_ack>=READ_QUORUM)
             return m;
        return null;
    }


    @Override
    public boolean containsKey(int key){
        /*ResponceMessage[] m;
        m=sendNewCommand(true,R_CONTAINS_KEY,key);
        if(m!=null)
         return (boolean)m[0].getResponce();
        else
         return false;*/
        return item_map.containsKey(key);
    }

    @Override
    public boolean removeKey(int key){
     //ResponceMessage[] m;
     //m=sendNewCommand(true,W_REMOVE_KEY,key);
     item_map.remove(key);
     return true;
     /*if(m!=null)
         return (boolean)m[0].getResponce();
     else
         return false;*/
    }

    @Override
    public boolean put(Product b){
     ResponceMessage[] m;
     m=sendNewCommand(true,W_PUT,b);
     item_map.put(Integer.parseInt(b.getId()),b);
     last_key_inserted=Integer.parseInt(b.getId());

     if(m!=null){
         //last_key_inserted=Integer.parseInt(b.getId());
         //return (boolean)m[0].getResponce();
         if((boolean)m[0].getResponce())
              System.out.println("V_TRUE");
         else
              System.out.println("FALSE");
     }//else
        // return false;

              return true;
    }

    @Override
    public Product get(int key){
     //ResponceMessage[] m;
     //m=sendNewCommand(true,R_GET,key);
     return item_map.get(key);
      /*if(m!=null)
         return (Product)m[0].getResponce();
      else
         return null;*/
    }

    @Override
    public int getLastKeyInserted(){
     return last_key_inserted;
    }

    @Override
    public void updatePrice(){
      //ResponceMessage[] m;
      //m=sendNewCommand(true,R_UPDATE_PRICE,null);
      for (Map.Entry pairs : item_map.entrySet()) {
            int key=(int)pairs.getKey();
            Product p=(Product)pairs.getValue();
            double price=p.getInitialPrice();
            item_map.remove(key);
            if(price>=5)
              p.setInitialPrice(price-5);
            else
              p.setInitialPrice(0);
            item_map.put(key, p);
        }
    }

    @Override
    public String generateJSONFromMap(){
        //ResponceMessage[] m;
        //m=sendNewCommand(true,R_GENERATE_JSON_FROM_MAP,null);
        String json="[";
        for (Map.Entry pairs : item_map.entrySet()) {
            Product p=(Product)pairs.getValue();
            json+=p.toString();
            json+=",";
        }

        if(json.endsWith(","))
            json = json.substring(0,json.length() - 1);
        json+="]";
        return json;
       /* if(m!=null)
         return (String)m[0].getResponce();
        else
         return null;*/
    }


    private void sendJMSMessageToTopicRepliche(String messageData) {
        context.createProducer().send(topicRepliche, messageData);
    }

    private void sendJMSMessageToTopicRepliche(ObjectMessage messageData) {
        context.createProducer().send(topicRepliche, messageData);
    }




    public AuctionFrontEnd(){
        final ConcurrentHashMap<Long,ResponceMessage> responce_list=new ConcurrentHashMap<>();
        this.listOfHashMap = new ArrayList<>();
        for(int i=0;i<2*NUM_REPLIC;i++){
            this.listOfHashMap.add(new ConcurrentHashMap<Long,ResponceMessage>());
        }
    }

}

前端消费者class:

/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package EJBData;

import Service.Command;
import Service.ResponceMessage;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.annotation.Resource;
import javax.ejb.Stateless;
import javax.jms.ConnectionFactory;
import javax.jms.JMSConsumer;
import javax.jms.JMSContext;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.ObjectMessage;
import javax.jms.Queue;


/**
 *
 * @author melix
 */

    @Stateless
    public class FrontEndConsumer {
        @Resource(lookup = "jms/__defaultConnectionFactory")
        private ConnectionFactory connectionFactory;
        @Resource(lookup = "QueueFrontEnd")
        private Queue queueFrontEnd;

        private JMSContext context;
        private JMSConsumer consumer;

        public void init(){
         context = connectionFactory.createContext();
         consumer = context.createConsumer(queueFrontEnd);
        }

        public void cleanQueue(){
           while(consumer.receive(1000)!=null){
            System.out.println("CANCELLO");
           }
        }

        public ResponceMessage consume(){
            Message m = consumer.receive(1000);
            if (m != null) {
               if(m instanceof ObjectMessage){
                   try {
                       Object object = ((ObjectMessage)m).getObject();
                       ResponceMessage mex=(ResponceMessage)object;  
                       System.out.println("RICEVO");
                       return (ResponceMessage)mex;
                   } catch (JMSException ex) {
                       Logger.getLogger(FrontEndConsumer.class.getName()).log(Level.SEVERE, null, ex);
                   }
               }
            }
            return null;
        }

    }

Answer 1

onMessage() 方法不需要等待 doOperation() 完成。 JMS API 启用异步通信，并在 JMS 提供程序的线程上调用 onMessage()。

如果 bean 部署在同一个应用程序服务器中，一个线程用于更多 bean？

A thread for more beans if beans are deployed in the same application server?

java

glassfish-4

jakarta-ee