当队列持久时，HornetQ Producer 变慢

Question

我已经尝试在 horntQ 中使用持久队列。我做了两个单独的例子（生产者，消费者）。我的消费者运行良好，但生产者花费了太多时间来完成发送消息。我已经分别运行以及一起运行。可能是什么问题呢？ 我的代码是：

public  class HornetProducer implements Runnable{

    Context ic = null;
    ConnectionFactory cf = null;
    Connection connection =  null;
    Queue queue = null;
    Session session = null;
    MessageProducer publisher =  null;
    TextMessage message = null;
    int messageSent=0;

     public synchronized static Context getInitialContext()throws javax.naming.NamingException {

            Properties p = new Properties( );
            p.put(Context.INITIAL_CONTEXT_FACTORY,"org.jnp.interfaces.NamingContextFactory");
            p.put(Context.URL_PKG_PREFIXES," org.jboss.naming:org.jnp.interfaces");
            p.put(Context.PROVIDER_URL, "jnp://localhosts:1099");

            return new javax.naming.InitialContext(p);
        }  

    public HornetProducer()throws Exception{            

        ic = getInitialContext();
        cf = (ConnectionFactory)ic.lookup("/ConnectionFactory");
        queue = (Queue)ic.lookup("queue/testQueue2");
        connection = cf.createConnection();
        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);        
        publisher = session.createProducer(queue);
        connection.start();

    }

    public void publish(){      
        try{        

            message = session.createTextMessage("Hello!");
            System.out.println("StartDate: "+new Date());

            for(int i=0;i<10000;i++){                   
                 messageSent++;              
                 publisher.send(message);                
            }
            System.out.println("EndDate: "+new Date());
        }catch(Exception e){
            System.out.println("Exception in Consume: "+ e.getMessage());
        }           
    }

    public void run(){
         publish();
    }

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

        new HornetProducer().publish();    
    }

}

Answer 1

您正在持续发送这些消息，并且是非事务性的。这意味着，发送的每条消息都必须单独完成。

这意味着对于您发送的每条消息，您必须与服务器进行一次网络往返，并等待它完成持久性，然后才能发送另一条消息。

如果您在这种情况下有多个生产者，hornetq 会批量处理这两个生产者，您将节省大量时间。 （即服务器将批处理许多写入请求）。

如果你想加快单个生产者的发送速度，你应该使用事务。

例如：

I - 将您的会话更改为已交易：

session = connection.createSession(true, Session.SESSION_TRANSACTIONED); 

II - 每 N 条消息提交一次：

   for(int i=0;i<10000;i++){                   
         messageSent++;              
         publisher.send(message);  
         if (messageSent % 1000 == 0) session.commit();              
    }
    session.commit();

您还可以禁用持久消息的同步。 （异步发送）。