首页 - 新闻 - 使用核心api消费后，HornetQ消息仍然保留在队列中

使用核心api消费后，HornetQ消息仍然保留在队列中

2023-09-24 03:49

| 我是HornetQ的新手，请多多包涵。首先让我告诉您我的要求：我需要一个消息排队中间件，该中间件可以在低延迟和持久性的不同进程之间传递大小约为1k的消息（即它应该在系统崩溃时幸免）。我将有多个进程写入同一队列，并且类似地有多个进程从同一队列读取。为此，我选择了HornetQ，因为它具有持久性传递消息的最佳评分。我目前使用Hornetq v2.2.2Final作为独立服务器。我能够使用核心api（ClientSession）成功创建持久/非持久队列，并成功将消息发布到队列（ClientProducer）。同样，我能够使用核心api（ClientConsumer）从队列中读取消息。问题出在此之后，当客户端读取消息时，消息仍保留在队列中，即，队列中的消息数保持不变。也许我弄错了，但我的印象是，一旦消息被消耗（读取+确认），就将其从队列中删除。再次。另外，我想告诉我，我已经尝试将非持久性队列与非持久性消息一起使用。但是问题仍然存在。我正在使用的生产者代码：

public class HQProducer implements Runnable {

    private ClientProducer producer;
    private boolean killme;
    private ClientSession session;
    private boolean durableMsg;

    public HQProducer(String host, int port, String address, String queueName,
            boolean deleteQ, boolean durable, boolean durableMsg, int pRate) {
        this.durableMsg = durableMsg;
        try {
            HashMap map = new HashMap();
            map.put(\"host\", host);
            map.put(\"port\", port);

            TransportConfiguration config = new TransportConfiguration(NettyConnectorFactory.class.getName(), map);

            ServerLocator locator = HornetQClient.createServerLocatorWithoutHA(config);

            ClientSessionFactory factory = locator.createSessionFactory();

            session = factory.createSession();

            if (queueExists(queueName)) {
                if (deleteQ) {
                    System.out.println(\"Deleting existing queue :: \" + queueName);
                    session.deleteQueue(queueName);
                    System.out.println(\"Creating queue :: \" + queueName);
                    session.createQueue(address, queueName, true);
                }
            } else {
                System.out.println(\"Creating new  queue :: \" + queueName);
                session.createQueue(address, queueName, durable);
            }
            producer = session.createProducer(SimpleString.toSimpleString(address), pRate);

            killme = false;
        } catch (Exception ex) {
            Logger.getLogger(HQTestProducer.class.getName()).log(Level.SEVERE, null, ex);
        }
    }

    @Override
    public void run() {
        long time = System.currentTimeMillis();
        int cnt = 0;
        long timediff;
        while (!killme) {
            try {
                ClientMessage message = session.createMessage(durableMsg);

                message.getBodyBuffer().writeString(\"Hello world\");

                producer.send(message);
                cnt++;
                timediff = ((System.currentTimeMillis() - time) / 1000);
                if (timediff >= 1) {
                    System.out.println(\"Producer tps :: \" + cnt);
                    cnt = 0;
                    time = System.currentTimeMillis();
                }
            } catch (HornetQException ex) {
                Logger.getLogger(HQProducer.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
        try {
            session.close();
        } catch (HornetQException ex) {
            Logger.getLogger(HQProducer.class.getName()).log(Level.SEVERE, null, ex);
        }
    }

    public void setKillMe(boolean killme) {
        this.killme = killme;
    }

    private boolean queueExists(String qname) {
        boolean res = false;
        try {
            //ClientSession.BindingQuery bq = session.bindingQuery(SimpleString.toSimpleString(qname));
            QueueQuery queueQuery = session.queueQuery(SimpleString.toSimpleString(qname));
            if (queueQuery.isExists()) {
                res = true;
            }
        } catch (HornetQException ex) {
            res = false;
        }
        return res;
    }
}

消费者的代码也是：

public class HQConsumer implements Runnable {

    private ClientSession session;
    private ClientConsumer consumer;
    private boolean killMe;

    public HQConsumer(String host, int port, String queueName, boolean browseOnly) {
        try {
            HashMap map = new HashMap();
            map.put(\"host\", host);
            map.put(\"port\", port);

            TransportConfiguration config = new TransportConfiguration(NettyConnectorFactory.class.getName(), map);

            ServerLocator locator = HornetQClient.createServerLocatorWithoutHA(config);

            ClientSessionFactory factory = locator.createSessionFactory();

            session = factory.createSession();

            session.start();

            consumer = session.createConsumer(queueName, \"\",0,-1,browseOnly);

            killMe = false;
        } catch (Exception ex) {
            Logger.getLogger(HQTestProducer.class.getName()).log(Level.SEVERE, null, ex);
        }
    }

    @Override
    public void run() {
        long time = System.currentTimeMillis();
        int cnt = 0;
        long timediff;
        while (!killMe) {
            try {
                ClientMessage msgReceived = consumer.receive();
                msgReceived.acknowledge();
                //System.out.println(\"message = \" + msgReceived.getBodyBuffer().readString());
                cnt++;
                timediff = ((System.currentTimeMillis() - time) / 1000);
                if (timediff >= 1) {
                    System.out.println(\"ConSumer tps :: \" + cnt);
                    cnt = 0;
                    time = System.currentTimeMillis();
                }
            } catch (HornetQException ex) {
                Logger.getLogger(HQConsumer.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
        try {
            session.close();
        } catch (HornetQException ex) {
            Logger.getLogger(HQConsumer.class.getName()).log(Level.SEVERE, null, ex);
        }
    }

    public void setKillMe(boolean killMe) {
        this.killMe = killMe;
    }
}

HornetQ服务器配置::



   ${data.dir:../data}/paging

   ${data.dir:../data}/bindings

   ${data.dir:../data}/journal

   10

   ${data.dir:../data}/large-messages

   
      
         org.hornetq.core.remoting.impl.netty.NettyConnectorFactory
         
         
      

      
         org.hornetq.core.remoting.impl.netty.NettyConnectorFactory
         
         
         
      
   

   
      
         org.hornetq.core.remoting.impl.netty.NettyAcceptorFactory
         
         
      

      
         org.hornetq.core.remoting.impl.netty.NettyAcceptorFactory
         
         
         
         
      
   

   
      
         
         
         
         
         
         
      
   

   
      
      
         jms.queue.DLQ
         jms.queue.ExpiryQueue
         0
         10485760       
         10
         BLOCK