activeMq的发布/订阅式的主题方式demo

创建一个maven项目，先导入相关依赖，对应的pom文件为：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>org.ssc</groupId>
    <artifactId>activemq_demo</artifactId>
    <version>1.0-SNAPSHOT</version>
<dependencies>
    <dependency>
        <groupId>org.apache.activemq</groupId>
        <artifactId>activemq-all</artifactId>
        <version>5.16.1</version>
    </dependency>
    <!-- 日志相关依赖 -->
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-api</artifactId>
        <version>1.6.1</version>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-log4j12</artifactId>
        <version>1.6.1</version>
    </dependency>
    <dependency>
        <groupId>log4j</groupId>
        <artifactId>log4j</artifactId>
        <version>1.2.16</version>
    </dependency>
</dependencies>
<build>
    <resources>
        <resource>
            <directory>src/main/java</directory>
            <includes>
                <include>**/*.xml</include>
            </includes>
        </resource>
    </resources>
</build>
</project>

创建一个消息生产者

package com.ssc.demo2;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class Producer {
    private static final String BROKER_URL="tcp://127.0.0.1:61616";
    public static void main(String[] args) throws JMSException {
        //连接工厂
        ConnectionFactory connectionFactory=null;
        //连接实例
       Connection connection=null;
       //收发的线程实例
        Session session=null;
        //消息发送目标地址
        Destination destination=null;
        try{
            //实例化连接工厂
            connectionFactory=new ActiveMQConnectionFactory("ssc", "ssc", BROKER_URL);
            //获取连接实例
            connection=connectionFactory.createConnection();
            //启动连接
            connection.start();
            //创建接收或发送的线程实例
            session =connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);
            //创建队列
            destination=session.createTopic("test.topic");
            //创建消息发布者
            MessageProducer messageProducer=session.createProducer(destination);
            //创建TestMessage消息
            TextMessage message=session.createTextMessage("你好测试，这是发布的消息");
            messageProducer.send(message);
            session.commit();
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            if(session!=null){
                session.close();
            }
            if(connection!=null){
                connection.close();
            }
        }
    }
}

查看activitemq控制台，控制台显示如下：

 创建两个订阅者，分别是：

package com.ssc.demo2;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class Customer01 {
    private static final String BROKER_URL="tcp://127.0.0.1:61616";
   public static void main(String[] args){
       //连接工厂
       ConnectionFactory connectionFactory=null;
       //连接实例
       Connection connection=null;
       //收发的线程信息
       Session session=null;
       //消息发送目标地址
       Destination destination=null;
       try{
           //创建连接工厂
           connectionFactory=new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_USER,ActiveMQConnectionFactory.DEFAULT_PASSWORD, BROKER_URL);
           //获取连接实例
           connection=connectionFactory.createConnection();
           //启动连接
           connection.start();
           //创建会话
           session =connection.createSession(Boolean.FALSE,Session.AUTO_ACKNOWLEDGE);
           //创建队列(返回一个消息目的地)
           destination=session.createTopic("test.topic");
           //创建消息订阅者
           MessageConsumer consumer=session.createConsumer(destination);
           //消息发布者添加监听器
           consumer.setMessageListener(new MessageListener() {
               public void onMessage(Message message) {
                   try {
                       System.out.println("订阅者01接收到的消息"+((TextMessage)message).getText());
                   }catch (JMSException e){
                       e.printStackTrace();
                   }

               }
           });
       } catch (JMSException e) {
           e.printStackTrace();
       }
   }
}

运行第一个消费者：

 创建并启动第二个消费者：

package com.ssc.demo2;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class Customer02 {
    private static final String BROKER_URL="tcp://127.0.0.1:61616";
    public static void main(String[] args){
        //连接工厂
        ConnectionFactory connectionFactory=null;
        //连接实例
        Connection connection=null;
        //收发的线程信息
        Session session=null;
        //消息发送目标地址
        Destination destination=null;
        try{
            //创建连接工厂
            connectionFactory=new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_USER,ActiveMQConnectionFactory.DEFAULT_PASSWORD, BROKER_URL);
            //获取连接实例
            connection=connectionFactory.createConnection();
            //启动连接
            connection.start();
            //创建会话
            session =connection.createSession(Boolean.FALSE,Session.AUTO_ACKNOWLEDGE);
            //创建队列(返回一个消息目的地)
            destination=session.createTopic("test1.topic");
            //创建消息订阅者
            MessageConsumer consumer=session.createConsumer(destination);
            //消息发布者添加监听器
            consumer.setMessageListener(new MessageListener() {
                public void onMessage(Message message) {
                    try {
                        System.out.println("订阅者02接收到的消息"+((TextMessage)message).getText());
                    }catch (JMSException e){
                        e.printStackTrace();
                    }

                }
            });
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

再次查看activemq控制台