• 官网：http://activemq.apache.org/
• 定义：发送者向消息服务器发送消息，将消息存储在多个队列/主题中topic在适当的时候，消息服务器会将消息转发给接受者。在此过程中，发送和接收是异步的，即发送不需要等待，发送和接收的生命周期之间没有必然的关系
• 特点：解耦、削峰、异步

• 下载到官网后，可以用来解压
• 修改activemq的conf目录中jetty.xml文件(访问外部设备)
  

• 普通启动：进入到activemq的bin目录中

  • ./activemq start
    

• 目的地Destination：队列(Queue)和主题(Topic)

• 第一步：创建Maven工程

• 第二步：引入配置（pom.xml）

  <dependency>     <groupId>org.apache.activemqgroupId>     <artifactId>activemq-allartifactId>
      <version>5.15.11version>
  dependency>
  
  <dependency>
      <groupId>org.apache.xbeangroupId>
      <artifactId>xbean-springartifactId>
      <version>4.15version>
  dependency>
  

• 特点
  • 1、每个消息只能有一个消费者，类似于1对1的关系。
  • 2、消息的生产者和消费者之间没有时间上的相关性。无论消费者在生产者发送消息的时候是否处于运行状态，消费者都可以提取消息。好比我们的发送短信，发送者发送后不见得接收者会即收即看
  • 3、消息被消费后队列中不会再存储，所以消费者不会消费到已经被消费掉的消息

public class JmsProduce { 
        

    public static final String ACTIVEMQ_URL = "tcp://192.168.37.130:61616";  //ActiveMQ地址
    public static final String QUEUE_NAME = "queue001";  //队列名称

    public static void main(String[] args) throws JMSException { 
        
        //1、创建连接工厂，按照给定的URL地址，采用默认的用户名和密码
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
        //2、通过连接工厂，获得连接connection并启动访问
        Connection connection = activeMQConnectionFactory.createConnection();
        connection.start();
        //3、创建会话Session（两个参数：transacted=是否开启事务，acknowledgeMode=签收模式）
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //4、创建目的地（具体是队列(queue)还是主题(top)）
        Queue queue = session.createQueue(QUEUE_NAME);
        //5、创建消息的生产者
        MessageProducer messageProducer = session.createProducer(queue);
        //6、创建消息并赋值
        TextMessage textMessage = session.createTextMessage();  //String类型的
        textMessage.setText("fzk");
        //7、发送消息到队列或主题
        messageProducer.send(textMessage);
        messageProducer.send(mapMessage);
        //8、关闭资源
        messageProducer.close();
        session.close();
        connection.close();

        System.out.println("消息发送完毕");
    }
}

• 控制台

• 控制台说明

  • Name=队列名称
  • Number Of Pending Messages=等待消费的消息，这个是未出队列的数量，公式=总接收数-总出队列数
  • Number Of Consumers=消费者数量，消费者端的消费者数量。
  • Messages Enqueued=进队消息数，进队列的总消息量，包括出队列的。这个数只增不减。
  • Messages Dequeued=出队消息数，可以理解为是消费者消费掉的数量。

• 订阅者或接收者抵用MessageConsumer的receive()方法来接收消息，receive方法在能接收到消息之前（或超时之前）将一直阻塞。

public class JmsConsumer { 
        

    public static final String ACTIVEMQ_URL = "tcp://192.168.37.130:61616";  //ActiveMQ地址
    public static final String QUEUE_NAME = "queue001";  //队列名称

    public static void main(String[] args) throws JMSException { 
        
        //1、创建连接工厂，按照给定的URL地址，采用默认的用户名和密码
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
        //2、通过连接工厂，获得连接connection并启动访问
        Connection connection = activeMQConnectionFactory.createConnection();
        connection.start();
        //3、创建会话Session（两个参数：transacted=是否开启事务，acknowledgeMode=签收模式）
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //4、创建目的地（具体是队列(queue)还是主题(top)）
        Queue queue = session.createQueue(QUEUE_NAME);
        //5、创建消息的生产者
        MessageConsumer messageConsumer = session.createConsumer(queue);
        //阻塞式
        while(true){ 
        
            //6、接收队列或主题的消息
            TextMessage textMessage = (TextMessage) messageConsumer.receive();
            if(textMessage != null){ 
        
                //7、获取消息中的数据
                String text = textMessage.getText();
                System.out.println(text);
            }else{ 
        
                break;
            }
        }
        //8、关闭资源
        messageConsumer.close();
        session.close();
        connection.close();
    }
}

• 订阅者或接收者通过MessageConsumer的setMessageListener(MessageListener listener)注册一个消息监听器，当消息到达之后，系统会自动调用监听器MessageListener的onMessage(Message message)方法。

• 特点

  • 1、生产者将消息发布到topic中，每个消息可以有多个消费者，属于1：N的关系
  • 2、生产者和消费者之间有时间上的相关性。订阅某一个主题的消费者只能消费自它订阅之后发布的消息
  • 3、生产者生产时，topic不保存消息它是无状态的不落地，假如无人订阅就去生产，那就是一条废消息，所以，一般先启动消费者再启动生产者

• 先启动订阅者再启动生产者,不然发送的消息是废消息

public class JmsProduceTopic { 
        

    public static final String ACTIVEMQ_URL = "tcp://192.168.37.130:61616";
    public static final String TOPIC_NAME = "topic001";

    public static void main(String[] args) throws JMSException { 
        
        //1、创建连接工厂，按照给定的URL地址，采用默认的用户名和密码
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
        //2、通过连接工厂，获得连接connection并启动访问
        Connection connection = activeMQConnectionFactory.createConnection();
        connection.start();
        //3、创建会话Session
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //4、创建目的地（具体是队列(queue)还是主题(top)）
        Topic topic = session.createTopic(TOPIC_NAME);
        //5、创建消息的生产者
        MessageProducer messageProducer = session.createProducer(topic);
        //6、创建消息并赋值
        TextMessage textMessage = session.createTextMessage();
        textMessage.setText("fzk");
        //7、发送消息到队列或主题
        messageProducer.send(textMessage);
        //8、关闭资源
        messageProducer.close();
        session.close();
        connection.close();

        System.out.println("消息发送完毕");
    }
}

public class JmsConsumerTopic { 
        

    public static final String ACTIVEMQ_URL = "tcp://192.168.37.130:61616";
    public static final String TOPIC_NAME = "topic001";

    public static void main(String[] args) throws JMSException { 
        
        //1、创建连接工厂，按照给定的URL地址，采用默认的用户名和密码
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
        //2、通过连接工厂，获得连接connection并启动访问
        Connection connection = activeMQConnectionFactory.createConnection();
        connection.start();
        //3、创建会话Session
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //4、创建目的地（具体是队列(queue)还是主题(top)）
        Queue queue = session.createQueue(QUEUE_NAME);
        //5、创建消息的生产者
        MessageConsumer messageConsumer = session.createConsumer(queue);
        //阻塞式
        while(true){ 
        
            //6、接收队列或主题的消息
            TextMessage textMessage = (TextMessage) messageConsumer.receive();
            if(textMessage != null){ 
        
                //7、获取消息中的数据
                String text = textMessage.getText();
                System.out.println(text);
            }else{ 
        
                break;
            }
        }
        //8、关闭资源
        messageConsumer.close();
        session.close();
        connection.close();
    }
}

public class JmsConsumerTopic { 
        

    public static final String ACTIVEMQ_URL = "tcp://192.168.37.130:61616";
    public static final String TOPIC_NAME = "topic001";

    public static void main(String[] args) throws JMSException, IOException { 
        
        System.out.println("消费者1");
        //1、创建连接工厂，按照给定的URL地址，采用默认的用户名和密码
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
        //2、通过连接工厂，获得连接connection并启动访问
        Connection connection = activeMQConnectionFactory.createConnection();
        connection.start();
        //3、创建会话Session
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //4、创建目的地（具体是队列(queue)还是主题(top)）
        Topic topic = session.createTopic(TOPIC_NAME);
        //5、创建消息的生产者
        MessageConsumer messageConsumer = session.createConsumer(topic);
        //6、使用异步监听式来获取消息
        messageConsumer.setMessageListener(new MessageListener() { 
        
            public void onMessage(Message message) { 
        
                if(message != null && message instanceof TextMessage){ 
        
                    //7、获取消息并通过消息获得数据
                    TextMessage textMessage = (TextMessage) message;
                    try { 
        
                        System.out.println(textMessage.getText());
                    } catch (JMSException e) { 
        
                        e.printStackTrace();
                    }
                }
            }
        });
        System.in.read();  //输入键盘任意键结束
        //8、关闭资源
        messageConsumer.close();
        session.close();
        connection.close();
    }
}

• Java Message Service(Java消息服务是JavaEE中的一个技术)
  • Java消息服务指的是两个应用程序之间进行异步通信的API，它为标准协议和消息服务提供了一组通用接口，包括创建、发送、读取消息等，用于支持Java应用程序开发。在JavaEE中，当两个应用程序使用JMS进行通信时，它们之间不是直接相连的，而是通过一个共同的消息收发服务组件关联起来以达到解耦/异步削峰的效果

• 实现JMS接口和规范的消息中间件，也就是我们说的MQ服务器

• 消息生产者，创建和发送JMS消息的客户端应用

• 消息消费者，接收和处理JMS消息的客户端应用

• JMSDestination：消息发送的目的地，主要是指队列Queue和主题Topic
• JMSDeliveryMode：持久模式和非持久模式
  • 一条持久性的消息：应该被传送“一次仅仅一次”，这就意味着如果JMS提供者出现故障，该消息并不会丢失，它会在服务器恢复之后再次传递
  • 一条非持久的消息：最多会传递一次，这意味着服务器出现故障，该消息将会永远丢失
• JMSExpiration：过期时间（默认是永不过期）
  • 消息过期时间，等于Destination的send方法中的timeToLive值加上发送时刻的GMT时间值
  • 如果timeToLive值等于0，则JMSExpiration被设为0，表示该消息永不过期
  • 如果发送后，在消息过期时间之后还没有被发送到目的地，则该消息被清除
• JMSPriority：消息优先级
  • 从0-9十个级别，0-4是普通消息5-9是加急消息
  • JMS不要求MQ严格按照这十个优先级发送消息但必须保证加急消息要先于普通消息到达。默认是4级
• JMSMessageID：唯一标识每个消息的标识由MQ产生