目录

前言

发送方式

P2P(queue）

Pub/Sub(topic)

主要步骤

具体实现

添加maven依赖

P2P方式

Pub/Sub方式(topic)

配置安全认证

前言

这里对activemq消息队列学习使用过程做一个简单得记录，也算作是加深记忆，不对之处还望赐教。

发送方式

• P2P(queue）

P2P （点对点）消息域使用 queue 作为 Destination，消息可以被同步或异步的发送和接收，每个消息只会给一个 Consumer 传送一次。

Consumer 可以使用 MessageConsumer.receive() 同步地接收消息，也可以通过使用MessageConsumer.setMessageListener() 注册一个 MessageListener 实现异步接收。

多个 Consumer 可以注册到同一个 queue 上，但一个消息只能被一个 Consumer 所接收，然后由该 Consumer 来确认消息。并且在这种情况下，Provider 对所有注册的 Consumer 以轮询的方式发送消息。

• Pub/Sub(topic)

Pub/Sub（发布/订阅，Publish/Subscribe）消息域使用 topic 作为 Destination，发布者向 topic 发送消息，订阅者注册接收来自 topic 的消息。发送到 topic 的任何消息都将自动传递给所有订阅者。接收方式（同步和异步）与 P2P 域相同。

除非显式指定，否则 topic 不会为订阅者保留消息。当然，这可以通过持久化（Durable）订阅来实现消息的保存。这种情况下，当订阅者与 Provider 断开时，Provider 会为它存储消息。当持久化订阅者重新连接时，将会受到所有的断连期间未消费的消息。

主要步骤

1. 获取工厂连接

2. 获取连接

3. 启动连接

4. 创建会话Session

5. 创建目的地

6. 创建消费者或者生产者

7. 发送消息或者监听消息

8. 关闭producer、consumer、session、connection等

具体实现

• 添加maven依赖

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

• P2P方式

     1）生产者(producer)

package com.activemq.demo;import org.apache.activemq.ActiveMQConnectionFactory;import javax.jms.*;public class MQProducer {public static void main(String[] args) {try {produceMsg("activemq.test","activemq 测试通信！");} catch (JMSException e) {e.printStackTrace();}}public static void produceMsg(String queue,String msg) throws JMSException {//配置地址String url = "tcp://192.168.239.128:61616";//创建连接工厂ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(url);//创建连接Connection connection = activeMQConnectionFactory.createConnection();//启动连接connection.start();//创建会话Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);//创建目的地Queue sessionQueue = session.createQueue(queue);//创建生产者MessageProducer producer = session.createProducer(sessionQueue);//创建消息TextMessage textMessage = session.createTextMessage(msg);//发送消息producer.send(textMessage);//关闭producer、session、connectionproducer.close();session.close();connection.close();}
}

2）消费者(consumer)

package com.activemq.demo;import org.apache.activemq.ActiveMQConnectionFactory;import javax.jms.*;public class MQConsumer {public static void main(String[] args) {try {consumeMsg("activemq.test");} catch (JMSException e) {e.printStackTrace();}}public static  void consumeMsg(String queue) throws JMSException {//通信地址String url  = "tcp://192.168.239.128:61616";//创建连接工厂ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(url);//创建工厂Connection connection = activeMQConnectionFactory.createConnection();//启动连接connection.start();//创建会话Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);//创建目的地Queue sessionQueue = session.createQueue(queue);//创建消费者MessageConsumer consumer = session.createConsumer(sessionQueue);//创建监听器consumer.setMessageListener(message->{if(message instanceof TextMessage){TextMessage textMessage = (TextMessage) message;try {System.out.println(textMessage.getText());} catch (JMSException e) {e.printStackTrace();}}});//关闭consumer、session、connection/*consumer.close();session.close();connection.close();*/}
}

注意：消费者这里要一直处于监听状态，不要直接关闭consumer、session等等

3）测试

先启动 MQproducer ，打开管理UI http://192.168.239.128:8161 ，我们会看到一条未被消费的信息

再启动 MQConsumer，再观察消费的信息变化情况，本地控制台也有打印信息的

• Pub/Sub方式(topic)

1)生产者(producer)

package com.activemq.demo;import org.apache.activemq.ActiveMQConnectionFactory;import javax.jms.*;public class MQProducer {public static void main(String[] args) {try {produceMsg("activemq.topic","activemq 测试通信！");} catch (JMSException e) {e.printStackTrace();}}public static void produceMsg(String topic,String msg) throws JMSException {//配置地址String url = "tcp://192.168.239.128:61616";//创建连接工厂ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(url);//创建连接Connection connection = activeMQConnectionFactory.createConnection();//启动连接connection.start();//创建会话Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);//创建目的地
//        Queue sessionQueue = session.createQueue(queue);//创建目的地Topic sessionTopic = session.createTopic(topic);//创建生产者MessageProducer producer = session.createProducer(sessionTopic);//创建消息TextMessage textMessage = session.createTextMessage(msg);//发送消息producer.send(textMessage);//关闭producer、session、connectionproducer.close();session.close();connection.close();}
}

2)消费者(consumer)

package com.activemq.demo;import org.apache.activemq.ActiveMQConnectionFactory;import javax.jms.*;public class MQConsumer {public static void main(String[] args) {try {consumeMsg("activemq.topic");} catch (JMSException e) {e.printStackTrace();}}public static  void consumeMsg(String topic) throws JMSException {//通信地址String url  = "tcp://192.168.239.128:61616";//创建连接工厂ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(url);//创建工厂Connection connection = activeMQConnectionFactory.createConnection();//启动连接connection.start();//创建会话Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);//创建目的地
//        Queue sessionQueue = session.createQueue(queue);//创建目的地Topic sessionTopic = session.createTopic(topic);//创建消费者MessageConsumer consumer = session.createConsumer(sessionTopic);//创建监听器consumer.setMessageListener(message->{if(message instanceof TextMessage){TextMessage textMessage = (TextMessage) message;try {System.out.println(textMessage.getText());} catch (JMSException e) {e.printStackTrace();}}});//关闭consumer、session、connection/*consumer.close();session.close();connection.close();*/}
}

总结：1. topic和queue在创建方式上大题相同，只是目的地创建方式不同

          2. queue的方式ActiveMq默认保留消息直到被消费，topic不会

          3. queue只能被一个消费者消费，多个消费者注册同一个queue也只会存在最后一个注册的消费者

配置安全认证

1)  登录activemq服务器，切换到配置目录。 cd /home/activemq/apache-activemq-5.15.11/conf

2）编辑配置文件。 vim activemq.xml 

3）添加内容如下。

<plugins><simpleAuthenticationPlugin><users><authenticationUser username="system" password="manager" groups="users,admins"/><authenticationUser username="user" password="password" groups="users"/><authenticationUser username="guest" password="password" groups="guests"/></users></simpleAuthenticationPlugin>
</plugins>

参考官网配置：http://activemq.apache.org/security

4）重新启动activemq。 service activemq  start

5）不修改代码测试

6）修改创建连接工厂参数 

  //创建连接工厂ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory("system","manager",url);

7）测试

总结：这里只是简单的身份验证，复杂的目的地控制和授权可以根据官网来进行配置