ActiveMQ，一个主题上的多个消费者会减慢生产者发送速率吗？

Question

这里的一些配置：非持久消费者、非持久消息、禁用的流控制、默认预取大小、优化确认=true、异步发送=true、使用jms连接ActiveMQ

例如

一个生产者、一个消费者，

Producer————Topic————consumer

生产者发送速率可以达到6k/s

但是，在这种情况下:一个生产者三个消费者，

                /——consumer

Producer——-Topic——-consumer

                \——consumer

生产者发送速率下降到4k/s

这是我的一些关键代码：

发件人类别:

public class sender {

    public Boolean durable=false;
    public String clientID=null;
    public Boolean transacted=false;
    public int ackMode=Session.AUTO_ACKNOWLEDGE;
    public int timeToLive=0;
    public String queuename = "";
    public int persistent = DeliveryMode.NON_PERSISTENT;

    public Connection createConnection(String user,String pwd,String url) throws JMSException, Exception {   
         ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user, pwd, url);
         connectionFactory.setDispatchAsync(true);
         //connectionFactory.setAlwaysSessionAsync(false);
         Connection connection = connectionFactory.createConnection();   
         if (durable && clientID!=null) {   
             connection.setClientID(clientID);   
         }   
         connection.start();   
         return connection;   
        }  

    public Session createSession(Connection connection) throws Exception {   
        Session session = connection.createSession(transacted, ackMode);   
        return session;   
       }   

    public MessageProducer createProducer(Session session) throws JMSException {   
        Queue destination = session.createQueue(queuename);   
        MessageProducer producer = session.createProducer(destination);   
        producer.setDeliveryMode(persistent);   

        if( timeToLive!=0 )   
            producer.setTimeToLive(timeToLive);   
        return producer;   
        }   

    public void onMessage(Message message) {   
         //process message   
         } 
}

sendmain方法：

public static void main(String[] args) throws JMSException, Exception {
        // TODO Auto-generated method stub
        sender s = new sender();
        s.persistent = DeliveryMode.NON_PERSISTENT;
        Connection c = s.createConnection("","","tcp://localhost:61616?jms.useAsyncSend=true");
        Session sess = s.createSession(c);
        Topic topic = sess.createTopic("topic.test");
        MessageProducer mp = sess.createProducer(topic);
        StringBuffer tmpsb=new StringBuffer();
        for (int j=0;j<1024;j++)
        {
        tmpsb.append("0");
        }
        Message m = sess.createTextMessage(tmpsb.toString());
        long pre=System.currentTimeMillis();
        for (int i=0;i<10000;i++)
        {
            mp.send(m);
        }
        long post=System.currentTimeMillis();
        mp.close();
        System.out.println("sendtime:"+(post-pre));
        System.out.println("sendrate:"[phone]/(float)(post-pre));
        System.out.println("timenow:"+(post));
    }

接收机类代码：

public class receiver implements MessageListener
{
    public  int receivetimes=0;
    public long pretime;

    public void onMessage(Message msg)
    {
        //TextMessage tm = (TextMessage) msg;
        try {
            if (receivetimes==0)
            {
                pretime=System.currentTimeMillis();
            }
            receivetimes+=1;
            if (receivetimes==10000)
            {
                long now=System.currentTimeMillis();
                System.out.println("time:"+(now-pretime)+"\nrecive rate:"+9999999/(float)(now-pretime));
                System.out.println("timenow:"+(now));
                receivetimes=0;
            }

        } catch(Throwable t) {
            t.printStackTrace();
        }
    }
}

接收器类代码在这里隐藏了一些方法，例如创建连接，创建会话或类似发送者类的东西。

接收器主要方法：

public static void main(String[] args) throws JMSException, Exception {
        // TODO Auto-generated method stub
        receiver s = new receiver();
        Connection c = s.createConnection("","","tcp://localhost:6151?jms.optimizeAcknowledge=true");
        Session sess = s.createSession(c);
        Topic destination  = sess.createTopic("topic.test");   
        MessageConsumer  consumer = sess.createConsumer(destination);  
        consumer.setMessageListener(new receiver());   
    }

每个消费者都在一个独立的过程中。我运行了三个消费者和一个生产者，然后我得到了一个表现不佳的结果。有人知道我为什么会得到这个吗？

Answer 1

正如@TimBish所说。问题是“生产者，消费者，同一台机器上的activemq服务器”。当分开时，问题永远不会出现。

严格地测试某些东西是非常重要的。。。。。。。