activeMq初识 - 2

Posted 2021-01-07 for-what

activeMq简单实例：

package com.gordon;


import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/**
 * 生产者
 * Created by gordon on 2018/9/8.
 */
public class Sender {

    public static final String userName = "admin";
    public static final String password = "admin";
    public static final String connect_url = "tcp://192.168.157.4:61616";
    public static final String QUEUE = "queue";
    public static final String TOPIC = "topic";

    public static void main(String[] args) throws JMSException {
//        testSender(QUEUE);
        testSender(TOPIC);// 测试广播模式先启动客户端
    }

    private static void testSender(String mode) throws JMSException {
        ConnectionFactory connectionFactory =
                new ActiveMQConnectionFactory(userName,password,connect_url);
        Connection connection = connectionFactory.createConnection();
        connection.start();
        Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
        TextMessage massage = session.createTextMessage();
//        Queue queue = session.createQueue(QUEUE);
//        Topic topic = session.createTopic(TOPIC);
        Destination destination = null;
        if(mode.equals(QUEUE)){
            destination = session.createQueue(mode);
            massage.setText("测试queue队列模式");
        }else if (mode.equals(TOPIC)){
            destination = session.createTopic(mode);
            massage.setText("测试topic广播模式");
        }
        MessageProducer producer = session.createProducer(destination);
        // 使用MessageProducer的serDeliveryMode方法为其设置持久化特性和非持久化特性（DeliveryMode）
        // producer.setDeliveryMode(DeliveryMode.PERSISTENT);将消息传递特性置为持久化
        producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);// 非持久化
        producer.send(massage);
        System.out.println("====生产者发送消息=====");
        if(connection != null){
            connection.close();
        }
        System.out.println("====生产者发送消息===结束==");
    }

}

View Code

package com.gordon;


import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;
import javax.jms.Destination;
import javax.jms.Queue;

/**
 * 消费者
 * Created by gordon on 2018/9/6.
 */
public class Receiver {

    public static void main(String[] args) throws JMSException {
//        testReceiver(Sender.QUEUE);
        testReceiver(Sender.TOPIC);
    }


    public static void testReceiver(String mode) throws JMSException {
        /* 1、建立ConnectionFactory工厂对象，需要填入用户名、密码、以及要连接的地址，
               注意不要包的路径，不要倒成spring里面的*/
        ConnectionFactory connectionFactory =
                new ActiveMQConnectionFactory(Sender.userName,Sender.password,Sender.connect_url);

        /* 2、通过ConnectionFactory工厂对象创建一个Connection连接，
             并且调用Connection.start()方法开启连接，默认是关闭的*/
        Connection connection = connectionFactory.createConnection();
        connection.start();

        /* 3、通过Connection对象创建Session会话(上下文环境对象)，用于接收消息。Connection可以创建一个或多个Session
             注意：会话是单线程的，所以消息是连续的，就是说消息是按照发送的顺序一个一个接收的
             参数1：是否启用事务(表示不开启事务)   参数2：接收模式(一般设置为自动接收)*/
        Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);

        /* 4、通过Session创建Destination对象，即一个客户端用来指定生产者目标 和 消费者来源 的对象
                Destination destination = session.createQueue("queue");
              PTP模式中：Destination被称为Queue，队列
              Pub/Sub模式中：Destination被称为Topic，主题
              在程序中可以使用多个Queue 和 Topic */
        //Queue queue1 = session.createQueue(mode);// 注意要和消息生产者的队列名一致
        Destination destination = null;
        if(mode.equals(Sender.QUEUE)){
            destination = session.createQueue(mode);
        }else if (mode.equals(Sender.TOPIC)){
            destination = session.createTopic(mode);
        }


        /* 5、通过Session对象创建消息的发送对接收对象(生产者或消费者,MessageProducer/MessageConsumer)*/
        MessageConsumer messageConsumer = session.createConsumer(destination);
        // 多次运行main方法，测试创建多个消费者接收topic消息
        System.out.println("messageConsumer : " + messageConsumer.hashCode());
        /* 6、使用JMS规范消息类型创建消息数据
               1. StreamMessage : Java原始值的数据流
               2. MapMessage : 一套名称-值对
               3.TextMessage : 一个字符串对象
               4.ObjectMessage : 一个序列化的Java对象
               5.BytesMessage : 一个未解释字节的数据流
             发送端使用MessageProducer.send()方法发送数据
             接收端使用MessageConsumer.receive()接收数据 */
        //System.out.println("AAAAAAAAAA");
        //TextMessage msg = (TextMessage) messageConsumer.receive();
        // receiver()同步方式，执行后当前一直挂起，等待接收消息，后面那句打印B的语句没有输出，
        // 知道发送者发送消息，这里接收到消息才会继续往后执行
        //System.out.println("BBBBBBBBBB");

        //System.out.println("===消费者接收===" + msg.getText());
        /* 7、最后一定要关闭connection连接,否则ActiveMQ不能释放资源，
              关闭一个Connection同样也关闭了Session、MessageProducer、MessageConsumer。

              但是，客户端一般不要关闭连接，除非特殊情况，这也是手动关闭(在页面搞个按钮点击关闭)。
              为什么？？很简单，如果客户端关闭，那如何接收消息？难道每次都要手动开启连接？？这也太麻烦了点。
              当然也有这种操作，毕竟少。我们都希望它是自动接收，所以客户端一般都一直保持连接
              */
//        if(connection != null){
//            connection.close();
//            System.out.println("===消费者接收===结束===");
//        }


        // 生成环境一般是用监听器方式，此方式是异步执行，同时生产者开启事务或者使用持久化(持久化可以设置文件大小)
        messageConsumer.setMessageListener(new CustomMessageListener());

    }

}

View Code

package com.gordon;

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

/**
 * 消息监听器，它会另起一个线程监听消息服务器
 * Created by gordon on 2018/9/8.
 */
public class CustomMessageListener implements MessageListener {
    @Override
    public void onMessage(Message message) {
        TextMessage textMessage = (TextMessage) message;
        try {
            String text = textMessage.getText();
            System.out.println(" ===== 消费者 接收消息 === " + text);
        } catch (JMSException e) {
            e.printStackTrace();
        }

    }
}

View Code

 

启动关闭：activeMq start / activeMq stop

activeMq端口：8161(web管理页面端口）、61616（activemq服务监控端口）

进入web控制台：http://192.168.157.4:8161

activeMq默认登录账号密码：admin admin

消息发布的两种方式：
　　topic：广播形式 

　　　　（1）每个消息可以有多个消费者
　　　　（2）生产者和消费者之间有时间上的相关性。订阅一个主题的消费者只能消费自它
　　　　 　　 订阅之后发布的消息。JMS 规范允许客户创建持久订阅，这在一定程度上放松了时间
　　　 　　　上的相关性要求。持久订阅允许消费者消费它在未处于激活状态时发送的消息。

　　ptp：点对点形式 

　　　　（1）每个消息只能有一个消费者
　　　　（2）消息的生产者和消费者之间没有时间上的相关性。无论消费者在生产者发送消
　　　　　　 息的时候是否处于运行状态，它都可以提取消息。

　　在点对点消息传递域中，目的地被称为队列（queue）；在发布/订阅消息传递域中，目的地被称为主题（topic）

activeMq web Queues控制台：
　　Number of Pending Message：等待消费的队列
　　Number of Consumers ：该队列被多少个消费者消费
　　Message Enqueued ：进入队列的消息总数量(只增不减)
　　Message Dequeued ：出了队列的消息总数量(即被消费了的消息数量)

 

参考：
 [activeMq基础系列文章]("https://www.cnblogs.com/HigginCui/category/955998.html")
 [activeMq消费者是如何接收消息的]("http://manzhizhen.iteye.com/category/318342")
 [activeMq性能优化系列文章]("https://blog.csdn.net/yinwenjie/article/details/50916518")
 [activeMq消息机制以及ACK机制详解]("http://shift-alt-ctrl.iteye.com/blog/2020182")
 [activeMq 常见问题解决]("https://blog.csdn.net/niwei3527/article/details/80607550")
 [activeMq面试题]("https://blog.csdn.net/belvine/article/details/79399798")