6

我已经尝试在 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();    
    }

}
4

1 回答 1

3

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

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

如果在这种情况下你有多个生产者，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();

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

于 2011-09-05T16:43:14.423 回答