ActiveMQ学习之java代码调用ActiveMQ队列

1、基本概念：

        ActiveMQ中共分为两种：queue和topic

                

            queue：在点对点消息传递域中，目的地被称为队列（一对一）

            topic：在发布订阅消息中，目的地被称为主题（一对多）

2、建立maven工程，并引入依赖，这里我建立的springboot项目，因此引入的依赖以下：

           依赖：            

<!--activemq-->
    <dependency>
      <groupId>org.apache.activemq</groupId>
      <artifactId>activemq-all</artifactId>
      <version>5.15.9</version>
    </dependency>

            建立main方法测试链接，消息提供者。

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/**
 * @ProjectName: springbootActiveMQ
 * @Package: cn.**.test
 * @Author: huat
 * @Date: 2020/1/2 17:04
 * @Version: 1.0
 */
public class ActiveMQTest {
    //url路径
    private static final String ACTRIVE_URL="tcp://192.168.44.135:61616";
    //队列名称
    private static final String QUEUE_NAME="queue01";
  

    public static void main(String[] args) {
        //一、建立链接工厂
        //若是帐号密码没有修改的话，帐号密码默认均为admin
        ActiveMQConnectionFactory activeMQConnectionFactory=new ActiveMQConnectionFactory(ACTRIVE_URL);
        //若是帐号密码修改的话
        //第一个参数为帐号，第二个为密码，第三个为请求的url
        //ActiveMQConnectionFactory activeMQConnectionFactory1=new ActiveMQConnectionFactory("admin","admin",ACTRIVE_URL);
        try {
            //二、经过链接工厂获取链接
            Connection connection = activeMQConnectionFactory.createConnection();
            connection.start();
            //三、建立session会话
            //里面会有两个参数，第一个为事物，第二个是签收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //四、建立目的地（具体是队列仍是主题）,这里是建立队列
            Queue queue=session.createQueue(QUEUE_NAME);
           
            //五、建立消息生产者
            MessageProducer messageProducer = session.createProducer(queue);
            //六、经过messageProducer生产三条消息发送到MQ消息队列中
            for (int i=0;i<3;i++){
                //七、建立消息
                TextMessage textMessage = session.createTextMessage("msg----->" + i);//建立一个文本消息
                //八、经过messageProducer发送给mq
                messageProducer.send(textMessage);
            }
            messageProducer.close();
            session.close();
            connection.close();
            System.out.println("消息发送成功");
        } catch (JMSException e) {
            e.printStackTrace();
        }

    }
}

            查看管理页面是否有消息

            建立main方法测试链接，消息消费者。（同步调用）

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/**
 * @ProjectName: springbootActiveMQ
 * @Package: cn.**.test
 * @Author: huat
 * @Date: 2020/1/3 8:47
 * @Version: 1.0
 */
public class ActiveMQConsumer {
    //url路径
    private static final String ACTRIVE_URL="tcp://192.168.44.135:61616";
    //队列名称
    private static final String QUEUE_NAME="queue01";

    public static void main(String[] args) {
        //一、建立链接工厂
        //若是帐号密码没有修改的话，帐号密码默认均为admin
        ActiveMQConnectionFactory activeMQConnectionFactory=new ActiveMQConnectionFactory(ACTRIVE_URL);
        //若是帐号密码修改的话
        //第一个参数为帐号，第二个为密码，第三个为请求的url
        //ActiveMQConnectionFactory activeMQConnectionFactory1=new ActiveMQConnectionFactory("admin","admin",ACTRIVE_URL);
        try {
            //二、经过链接工厂获取链接
            Connection connection = activeMQConnectionFactory.createConnection();
            connection.start();
            //三、建立session会话
            //里面会有两个参数，第一个为事物，第二个是签收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //四、这里接受的queue的名称要和发送者的一致
            Queue queue = session.createQueue(QUEUE_NAME);
            //五、建立消费者
            MessageConsumer consumer = session.createConsumer(queue);
            //六、获取消息
            while(true){
                //MessageConsumer 调用的receive方法为同步调用，在消息到达以前一直阻塞线程
                //用什么格式发送，这里就用什么格式接受
                //receive等待消息，不限制时间
                    TextMessage message=(TextMessage)consumer.receive();
                //receive带参数等待消息，限制时间，单位毫秒
                //TextMessage message=(TextMessage)consumer.receive(4000L);

                if(null != message){
                    System.out.println("接受的消息为------>"+message.getText());
                }else{
                    break;
                }
            }
            //七、闭资源
            consumer.close();
            session.close();
            connection.close();
        }catch (Exception e){
            e.printStackTrace();
        }
    }

}

            建立main方法测试链接，消息消费者。（异步调用）

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/**
 * @ProjectName: springbootActiveMQ
 * @Package: cn.**.test
 * @Author: huat
 * @Date: 2020/1/3 8:47
 * @Version: 1.0
 */
public class ActiveMQConsumer {
    //url路径
    private static final String ACTRIVE_URL="tcp://192.168.44.135:61616";
    //队列名称
    private static final String QUEUE_NAME="queue01";

    public static void main(String[] args) {
        //一、建立链接工厂
        //若是帐号密码没有修改的话，帐号密码默认均为admin
        ActiveMQConnectionFactory activeMQConnectionFactory=new ActiveMQConnectionFactory(ACTRIVE_URL);
        //若是帐号密码修改的话
        //第一个参数为帐号，第二个为密码，第三个为请求的url
        //ActiveMQConnectionFactory activeMQConnectionFactory1=new ActiveMQConnectionFactory("admin","admin",ACTRIVE_URL);
        try {
            //二、经过链接工厂获取链接
            Connection connection = activeMQConnectionFactory.createConnection();
            connection.start();
            //三、建立session会话
            //里面会有两个参数，第一个为事物，第二个是签收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //四、这里接受的queue的名称要和发送者的一致
            Queue queue = session.createQueue(QUEUE_NAME);
            //五、建立消费者
            MessageConsumer consumer = session.createConsumer(queue);
            //六、经过监听的方式消费消息
            consumer.setMessageListener(new MessageListener() {
                @Override
                public void onMessage(Message message) {
                    //若是message不等于null而且属于TextMessage类型（由于消息发送的类型是TextMessage，因此这里判断是不是这个类型）
                    if(null!=message&&message instanceof TextMessage){
                        TextMessage textMessage=(TextMessage)message;
                        try {
                            System.out.println(textMessage.getText());
                        } catch (JMSException e) {
                            e.printStackTrace();
                        }
                    }
                }
            });
            //七、保证控制台一直在运行
            System.in.read();
            //八、闭资源
            consumer.close();
            session.close();
            connection.close();
        }catch (Exception e){
            e.printStackTrace();
        }
    }

}

3、具体分析

            状况一：先启动生产者，再启动一号消费者，一号消费者能够消费所有消息

            状况二：先启动生产者，先启动一号消费者再启动二号消费者，一号消费者能够消费所有消息，二号消费者无消费

            状况三：先启动一号和二号两个消费者，再启动生产者，一号和二号消费者平均分配消息。