activitemq的点对点模型的队列方式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.demo1;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class Sender {
    //发送次数
    private static final int SEND_NUM=5;
    //tcp地址
    private static final String BROKER_url="tcp://127.0.0.1:61616";
    public static final String DESTINATION = "test1.mq.queue";

    public static void main(String[] args) throws Exception {
        Session session=null;
        ConnectionFactory connectionFactory=null;
        Connection connection=null;
        try {
            //第一步建立connectionFactory连接工厂，需要填入用户名，密码，以及要连接的地址
         connectionFactory=new ActiveMQConnectionFactory("ssc","ssc",BROKER_url );
       //第二步根据connectionFactory工厂对象给我们创建一个Connection连接，并且调用Connection的start方法开启连接
        connection=connectionFactory.createConnection();
        connection.start();
        //第三步通过Connection对象创建session会话（上下文环境对象），
        session=connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);
        //创建一个消息队列
        Destination destination=session.createQueue(DESTINATION);
        //创建消息制作者
        MessageProducer messageProducer=session.createProducer(destination);
        //创建持久化模式
        messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
        sendMessage(session,messageProducer);
        session.commit();

        }catch (Exception e){
            throw e;
        }finally {
            //关闭释放资源
            if (session!=null){
                session.close();
            }
            if(connection!=null){
                connection.close();
            }
        }
    }
    public static void sendMessage(Session session,MessageProducer messageProducer)throws Exception{
        for (int i=0;i<SEND_NUM;i++){
            String message="发送消息第"+(i+1)+"条";
            TextMessage text=session.createTextMessage(message);
            System.out.println(message);
            messageProducer.send(text);
        }
    }
}

运行服务发送者后，查看activemq控制台，会发现，等待处理的数据是5条，进入队列的数据也是5条，此时该消息未被消费者消费

服务发送者

package com.ssc.demo1;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class Consumer{
    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("ssc", "ssc", BROKER_URL);
            //获取连接实例
            connection=connectionFactory.createConnection();
            //启动连接
            connection.start();
            //创建接收和发送的线程实例
            session=connection.createSession(Boolean.FALSE,Session.AUTO_ACKNOWLEDGE);
            //创建一个消息队列
            destination=session.createQueue("test1.mq.queue");
            //创建消息消费者
            MessageConsumer messageConsumer=session.createConsumer(destination);
            //创建消息监听
            messageConsumer.setMessageListener(new MQListener());
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

消息监听器

package com.ssc.demo1;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;

public class MQListener implements MessageListener {

    public void onMessage(Message message) {
        try{
            System.out.println(((TextMessage)message).getText());
        }catch (JMSException e){
            e.printStackTrace();
        }
    }
}

运行消费者，后查看activemq控制台，会发现这5条消息会被消费