ActiveMQ（三十）--Message Dispatch高级特性2

一、批量消息确认

 

二、生产者流量控制

测试：

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class NonPerSender2 {
    public static void main(String[] args) throws JMSException {

        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.128:61616");
        Connection connection = connectionFactory.createConnection();
        connection.start();

        Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
        Destination destination = session.createTopic("MyTopic");
        for (int i = 0; i < 1; i++) {

            MessageProducer producer = session.createProducer(destination);

            for (int j = 0; j < 1; j++) {
                TextMessage message = session.createTextMessage(i + "===message222--" + j);
                producer.send(message);
            }

            session.commit();
        }
        session.close();
        connection.close();
    }
}

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class NoPerReceiver2 {
    public static void main(String[] args) throws JMSException {

        ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://192.168.25.128:61616");

        for (int i = 0; i < 1; i++) {
            Thread t = new MyT(factory);
            t.start();
        }
    }
}

class MyT extends Thread {
    private ConnectionFactory connectionFactory = null;

    public MyT(ConnectionFactory connectionFactory) {
        this.connectionFactory = connectionFactory;
    }

    @Override
    public void run() {
        try {
            Connection connection = connectionFactory.createConnection();
            connection.start();

            Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
            Destination destination = session.createTopic("MyTopic");

            for (int i = 0; i < 1; i++) {

                final MessageConsumer consumer = session.createConsumer(destination);
                consumer.setMessageListener(new MessageListener() {
                    @Override
                    public void onMessage(Message message) {
                        TextMessage txtMsg = (TextMessage) message;
                        try {
                            System.out.println(consumer + "收到消息：" + txtMsg.getText());
                            session.commit();
                        } catch (JMSException e) {
                            e.printStackTrace();
                        }
                    }
                });
            }
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

先启动消费者，后启动生产者。可以一条一条的发送和接收。

修改生产者的发送数量。

生产者会阻塞在那里。并且，要正常运行的话，要重启mq。

如果想每次都能发3条怎么办呢？

这个时候，没有事务。就不是批量提交了。而是一条一条的发。只要单条消息不超过10b，就能发送多条。