消息队列--ActiveMQ

JMS（Java Message Service，Java消息服务）是一组Java应用程序接口(Java API)，它提供创建、发送、接收、读取消息的服务。Java消息服务支持两种消息模型：Point-to-Point消息(P2P)和发布订阅消息（Publish Subscribe messaging，简称Pub/Sub）。

消息队列厂商只需要实现这些接口即可，与JDBC的实现过程是类似的（JDBC也是提供一组接口，然后MySQL有MySQL的实现，Oracle有Oracle的实现...）。

 

首先去ActiveMQ官网下载，我目前版本是 ActiveMQ 5.15.0 Release

 

然后解压，我解压到D盘

 

bin目录用来启动MQ，conf是配置文件，那个jar包是MQ的包，一会要用。

打开eclipse，新建一个Java工程，把这个包复制到工程，然后build path。

然后新建两个类，一个作为生产者，一个作为消费者。

生产者：

package mq.helloworld;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

public class Sender {

    
    public static void main(String[] args) throws Exception {
        
        //1. 建立一个ConnectionFactory. tcp://0.0.0.0:61616
        String userName = ActiveMQConnectionFactory.DEFAULT_USER;
        String password = ActiveMQConnectionFactory.DEFAULT_PASSWORD;
        String brokerURL = "tcp://localhost:61616";
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(userName, password, brokerURL);
        
        //2. 通过ConnectionFactory建立一个Connection连接，并且调用start方法开启
        Connection connection = connectionFactory.createConnection();
        connection.start();
        
        //3. 通过Connection创建Session，用于接收消息[第一个参数：是否启用事务；第二个参数：设置签收模式]
//        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);//手工签收--常用
        
        //4. 通过Session创建Destination对象
        Destination destination = session.createQueue("foodQueue");
        
        //5. 通过Session创建发送或接受对象
        MessageProducer messageProducer = session.createProducer(destination);
        
        //6. 设置持久化特性或非持久化特性
        messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
        
        //7. 使用JMS规范里面消息类型之一 TextMessage 来创建数据，用MessageProducer来发送
        for (int i = 0; i < 5; i++) {
            TextMessage message = session.createTextMessage();
            message.setText("大萝卜" + i);
            messageProducer.send(message);
            System.out.println("生产者：" + message.getText());
        }
        
        //使用事务要手动提交
        //session.commit();
        
        //8. 关闭连接
        connection.close();
        
    }// main
}

 

 消费者：

package mq.helloworld;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

public class Receiver {

public static void main(String[] args) throws Exception {
        
        //1. 建立一个ConnectionFactory. tcp://0.0.0.0:61616
        String userName = ActiveMQConnectionFactory.DEFAULT_USER;
        String password = ActiveMQConnectionFactory.DEFAULT_PASSWORD;
        String brokerURL = "tcp://localhost:61616";
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(userName, password, brokerURL);
        
        //2. 通过ConnectionFactory建立一个Connection连接，并且调用start方法开启
        Connection connection = connectionFactory.createConnection();
        connection.start();
        
        //3. 通过Connection创建Session，用于接收消息[第一个参数：是否启用事务；第二个参数：设置签收模式]
//        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
        
        //4. 通过Session创建Destination对象
        Destination destination = session.createQueue("foodQueue");
        
        //5. 通过Session创建发送或接受对象
        MessageConsumer messageConsumer = session.createConsumer(destination);
        
        //7. 使用JMS规范里面消息类型之一 TextMessage 来创建数据
        while (true) {
            TextMessage message = (TextMessage) messageConsumer.receive();
            if(message == null) break;
            message.acknowledge();    //手动签收
            System.out.println("消费者：" + message.getText());
        }
        
        //8. 关闭连接
        connection.close();
        
    }// main
}

 先启动MQ

 

在浏览器里输入http://localhost:8161/admin/

出现这个界面就ok。

然后运行Sender。

这里是控制台打印的，是否真的发出去了，看MQ的管理界面。

ok。然后运行receiver

 

好了，五个大萝卜已经吃了，看管理界面的变化。

注意：如果没有【手动签收】，MQ是不会认为客户端已经消费了的

message.acknowledge();    //手动签收

 

到此，一个hello 完成