一、创建Maven项目并引入依赖

<dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-all</artifactId>
    <version>5.16.2</version>
</dependency>

二、生产者程序

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/**
 * @author IT00ZYQ
 * @date 2021/6/12 22:34
 **/
public class Producer {
    public static void main(String[] args) throws JMSException, InterruptedException {
        // 1、创建连接工厂
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(
                ActiveMQConnectionFactory.DEFAULT_USER,
                ActiveMQConnectionFactory.DEFAULT_PASSWORD,
                "tcp://localhost:61616"
        );

        // 2、创建连接
        Connection connection = factory.createConnection();

        // 3. 创建Session(事务，确认)
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        // 4. 创建用于发送消息的队列
        Queue queue = session.createQueue("test-queue");

        // 5. 创建生产者
        MessageProducer producer = session.createProducer(queue);
        
        // 6. 创建消息并发送
        int i = 1;
        while (true) {
            System.out.println("发送消息：" + i);
            TextMessage message = session.createTextMessage("text-message -> " + i);
            producer.send(message);
            // 每3秒生产一个消息
            Thread.sleep(3000);
            i++;
        }

        // 6. 关闭连接
//        connection.close();
    }
}

三、消费者程序

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/**
 * @author IT00ZYQ
 * @date 2021/6/12 22:50
 **/
public class Consumer {
    public static void main(String[] args) throws JMSException, InterruptedException {
        // 1、创建连接工厂
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(
                ActiveMQConnectionFactory.DEFAULT_USER,
                ActiveMQConnectionFactory.DEFAULT_PASSWORD,
                "tcp://localhost:61616"
        );

        // 2、创建连接并开启
        Connection connection = factory.createConnection();
        connection.start();

        // 3. 创建Session(事务，确认)
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        // 4. 指定获取消息的队列
        Destination queue = session.createQueue("test-queue");

        // 5. 创建消费者
        MessageConsumer consumer = session.createConsumer(queue);

        // 6. 消费消息并输出
        while (true) {
            TextMessage message = (TextMessage)consumer.receive();
            System.out.println("消费消息：" + message.getText());
            // 每2秒消费一次
            Thread.sleep(2000);
        }

        // 7. 关闭连接
//        connection.close();
    }
}

四、ActiveMQ运行情况