ActiveMq简单的订阅发布（消息重发）

Session 中的四种type

• AUTO_ACKNOWLEDGE=1 自动确认
• CLIENT_ACKNOWLEDGE=2 客户端手动确认
• DUPS_OK_ACKNOWLEDGE=3 自动批量确认
• SESSION_TRANSACTED =0 事务提交并确认

public class ActiveMQTest {

    // 编写消息发送方 (生产者)
    @Test
    public void test1() throws JMSException {
        //创建连接工厂对象
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://ip地址:61616");
        //从工厂中获取一个连接对象
        Connection connection = connectionFactory.createConnection();
        //连接MQ服务
        connection.start();
        //获取session
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //通过session对象常见Topic
        Topic topic = session.createTopic("MainTopic");
        //通过session对象创建消息的发送者
        MessageProducer producer = session.createProducer(topic);
        //通过session 创建消息对象
        TextMessage message = session.createTextMessage("ping111");
        //发送消息
        producer.send(message);
        //关闭相关资源
        producer.close();
        session.close();
        connection.close();

    }

    // 编写消息接受方 (消费者)
    @Test
    public void test2() throws JMSException {
        //创建连接工厂对象
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://ip地址:61616");
        //从工厂中获取一个连接对象
        Connection connection = connectionFactory.createConnection();
        //连接MQ服务
        connection.start();
        //获取session
        final Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
        //通过session对象常见Topic
        Topic topic = session.createTopic("MainTopic");
        //通过session对象创建消息的发送者
        MessageConsumer consumer = session.createConsumer(topic);
        //指定消息监听器
        //当监听的topic中存在消息，会自动执行
        consumer.setMessageListener(message -> {
            TextMessage textMessage = (TextMessage) message;
            try {
                if (textMessage.getText().equals("ping")){
                    System.err.println("消费者接受到====>>>>> ： " + textMessage.getText());
                    //客户端手动应答
                    message.acknowledge();
                }else {
                    System.err.println("消息处理失败了。。。");
                    //通知mq进行消息重发，最多重发6次
                    session.recover();
                    //模拟消息处理失败
                    int i = 1/0;
                }
            } catch (JMSException e) {
                e.printStackTrace();
            }
        });
       while (true){

       }
    }

}