ActiveMQ_Java实现生产者和消费者操作队列

<dependency>
  <groupId>org.apache.activemq</groupId>
  <artifactId>activemq-all</artifactId>
  <version>5.15.5</version>
</dependency>
<dependency>
  <groupId>org.apache.xbean</groupId>
  <artifactId>xbean-spring</artifactId>
  <version>4.10</version>
</dependency>

public class JmsProducer {
    private static final String BROKER_URL = "tcp://192.168.229.129:61616";
    private static final String QUEUE_NAME = "queue01";
    private static final String TEXT_MESSAGE = "message";

    public static void main(String[] args) throws JMSException {
        // 1、创建连接工厂ConnectionFactory
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(BROKER_URL);
        // 2、通过连接工厂获取连接对象
        Connection connection = connectionFactory.createConnection();
        // 3、启动JMS连接
        connection.start();
        // 4、通过连接对象创建JMS Session对象,第一个参数为是否开启事务,第二个参数是消息的签收方式
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 5、创建Destination (Destination是 Queue/Topic的父接口)
        Queue queue = session.createQueue(QUEUE_NAME);
        // 6、创建消息的生产者,参数是Destination,也就是消息的生产者会将消息推送到哪里.
        MessageProducer producer = session.createProducer(queue);
        // 7、创建消息,创建三个消息发送至队列
        for (int i = 1; i < 4; i++) {
            TextMessage textMessage = session.createTextMessage(TEXT_MESSAGE + "0" + i);
            producer.send(textMessage);
        }
        // 8、释放资源,先开后关,后开先关
        producer.close();
        session.close();
        connection.close();
        System.out.println("消息成功发送至队列");
    }
}

public class JmsConsumer {
    private static final String BROKER_URL = "tcp://192.168.229.129:61616";
    private static final String QUEUE_NAME = "queue01";
    private static final String TEXT_MESSAGE = "message";

    public static void main(String[] args) throws JMSException {
        // 1、创建连接工厂对象
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(BROKER_URL);
        // 2、通过连接工厂获取连接对象
        Connection connection = connectionFactory.createConnection();
        // 3、启动JMS连接
        connection.start();
        // 4、通过connection对象获取JMS Session对象
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 5、创建Destination对象(Queue/Topic的父接口是Destination)
        Queue queue = session.createQueue(QUEUE_NAME);
        // 6、创建消费者对象
        MessageConsumer consumer = session.createConsumer(queue);
        // 7、消费者消费消息
        while (true) {
            // 时间为毫秒值,如果过了10s之后,消费者没有接收到新的消息,那么消费者自动和队列断开
            TextMessage textMessage = (TextMessage) consumer.receive(10*1000);
            /*TextMessage textMessage = (TextMessage) consumer.receive();*/
            if (textMessage != null) {
                String text = textMessage.getText();
                if (text != null) {
                    System.out.println("接收到的消息是:" + text);
                }
            } else {
                break;
            }
        }
        // 8、释放资源
        consumer.close();
        session.close();
        connection.close();
    }
}

// 一直接收消息,消费者和队列不会断开连接
TextMessage textMessage = (TextMessage) consumer.receive();
// 超过了时间(毫秒值),消费者将和队列自动断开连接
TextMessage textMessage = (TextMessage) consumer.receive(10*1000);

// 消费者通过消息监听的方式消费消息
consumer.setMessageListener(new MessageListener() {
	@Override
	public void onMessage(Message message) {
		if (message != null && message instanceof TextMessage) {
			TextMessage textMessage = (TextMessage) message;
			try {
				System.out.println("接收到的消息是:" + textMessage.getText());
			} catch (JMSException e) {
				e.printStackTrace();
			}
		}
	}
});
// 保证在连接到MQ之前,控制台不灭,也就是消费到了消息之后才去释放资源,Dubbo中也有
System.in.read();