JMS-中间件MQ-activeMQ入门

一、MQ(Message Queue)概念

1.1 MQ的技术维度

  • api发送和接收
  • MQ高可用
  • MQ集群和容错配置
  • MQ持久化
    • redis
  • 延时发送/定时投递
  • 签收机制
  • spring整合

1.2 作用

  • 解耦：系统间接口耦合严重
    • 当新的模块接进来时，可以做到代码改动最小
  • 削峰：面对大流量并发时，容易被冲垮
    • 设置流量缓冲池，后端系统按照自身吞吐能力进行消费
  • 异步：等待同步存在性能问题
    • 强弱依赖梳理能将不关键调用链路的操作异步化，提升系统的吞吐能力

1.3 定义

image

  • 面向消息的中间件(message-oriented middleware)MOM
    提供消息传递和消息排队模型在分布式环境下提供应用的解耦、弹性伸缩、冗余存储、流量削峰、异步通信、数据同步
  • 过程：
    • 发送方——>消息服务器，消息服务器将消息存放在：队列/主题，消息服务器——>接收方。
    • 发送和接收是异步的
    • 在pub/sub模式下，可以完成一对多，一个消息多个接收方。

二、简单操作

  • 安装过程略....
  • 解压命令：tar -zxvf ./apache-activemq-5.16.2-bin.tar.gz -C /指定目录
  • 复制到自己的目录：sudo cp -r apache-activemq-5.16.2/ ~/myactiveMQ/
  • 启动文件目录：/home/dj/myactiveMQ/apache-activemq-5.16.2/bin
总用量 144
-rwxr-xr-x. 1 root root 26692 8月   8 10:35 activemq
-rwxr-xr-x. 1 root root  6189 8月   8 10:35 activemq-diag
-rw-r--r--. 1 root root 16020 8月   8 10:35 activemq.jar
-rw-r--r--. 1 root root  5597 8月   8 10:35 env
drwxr-xr-x. 2 root root    78 8月   8 10:35 linux-x86-32
drwxr-xr-x. 2 root root    78 8月   8 10:35 linux-x86-64
drwxr-xr-x. 2 root root    82 8月   8 10:35 macosx
-rw-r--r--. 1 root root 83820 8月   8 10:35 wrapper.jar
  • 启动命令：./activemq start
  • activeMQ的默认进程端口：61616
  • 检查是否开启命令：
    • lsof -i:61616
    java    3611 root  137u  IPv6  45437      0t0  TCP *:61616 (LISTEN)
    • netstat -anp|grep 61616
    tcp6       0      0 :::61616                :::*                    LISTEN      3611/java
  • 关闭命令：./activemq stop
  • 带日志的启动命令：./activemq start > /myactiveMQ/run_activemq.log
  • activeMQ控制台： 
安装iptables-service工具：   yum install iptables-service
启用：                      systemctl enable iptables
打开：                      systemctl start iptables
查看防火墙状态：             service iptables status
关闭防火墙：                service iptables stop
打开防火墙：                service iptables start

三、java编码MQ的api

3.1 依赖：

<!-- activemq的依赖 -->
<dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-all</artifactId>
    <version>5.16.2</version>
</dependency>
<!-- activemq整合spring的基础包 -->
<dependency>
    <groupId>org.apache.xbean</groupId>
    <artifactId>xbean-spring</artifactId>
    <version>4.20</version>
</dependency>

3.2 执行原理

Destination(目的地):

  • Queue(队列)：一对一
  • Topic(主题)：一对多

image

3.3 Queue-测试代码实例

3.3.1 生产者编码

public class JmsProducer {
    //url地址
    public static final String MY_ACTIVEMQ_URL = "tcp://192.168.10.100:61616";
    public static final String MY_QUEUE_NAME = "queue01";

    public static void main(String[] args) throws JMSException {
        //1 创建Connection连接工厂
        ActiveMQConnectionFactory acf = new ActiveMQConnectionFactory(MY_ACTIVEMQ_URL);
        //2 使用连接工厂创建连接并启动
        Connection connection = acf.createConnection();
        connection.start();
        //3 创建会话Session,两个参数：第一个事务/第二个签收
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //4 创建目的地
        Queue queue = session.createQueue(MY_QUEUE_NAME);
        //5 创建消息生产者
        MessageProducer messageProducer = session.createProducer(queue);

        for (int i = 1; i <= 3; i++){
            //6 创建消息
            TextMessage message = session.createTextMessage("消息msg--->" + i);
            //7 messageProducer将产生的消息发送到MQ队列中
            messageProducer.send(message);
        }
        //8 关闭资源
        messageProducer.close();
        session.close();
        connection.close();

        //测试程序完成
        System.out.println("*****程序发布成功");
    }
}
  • web控制台效果：
    image

3.3.2 消费者编码

public class JmsConsumer {

    //url地址
    public static final String MY_ACTIVEMQ_URL = "tcp://192.168.10.100:61616";
    public static final String MY_QUEUE_NAME = "queue01";

    public static void main(String[] args) throws JMSException {
        //1 创建Connection连接工厂
        ActiveMQConnectionFactory acf = new ActiveMQConnectionFactory(MY_ACTIVEMQ_URL);
        //2 使用连接工厂创建连接并启动
        Connection connection = acf.createConnection();
        connection.start();
        //3 创建会话Session,两个参数：第一个事务/第二个签收
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //4 创建目的地
        Queue queue = session.createQueue(MY_QUEUE_NAME);
        //5 创建消费者
        MessageConsumer messageConsumer = session.createConsumer(queue);

        while (true){
            //6 接收消息
            TextMessage message = (TextMessage) messageConsumer.receive();
            //TextMessage message = (TextMessage) messageConsumer.receive(4000L);
            //有时间的receive方法，设定的时间一过，自动关闭
            if (message != null){
                System.out.println("接收消息===>"+message.getText());
            }else {
                System.out.println("!!!!接收完毕!!!!!");
                break;
            }
        }
        //7 关闭资源
        messageConsumer.close();
        session.close();
        connection.close();
    }
}
  • 控制台效果：
    image

3.3.3 消费者编码-监听器

//5 创建消费者
MessageConsumer messageConsumer = session.createConsumer(queue);

/*
* 方法一：同步阻塞
* */
//  while (true){
//      //6 接收消息
//      TextMessage message = (TextMessage) messageConsumer.receive();
//      if (message != null){
//          System.out.println("接收消息===>"+message.getText());
//      }else {
//          System.out.println("!!!!接收完毕!!!!!");
//          break;
//      }
//  }

/*
* 方法二：设置监听器
* */
messageConsumer.setMessageListener(new MessageListener() {
    @Override
    public void onMessage(Message message) {
        if (message != null && message instanceof TextMessage){
            TextMessage textMessage = (TextMessage) message;
            try {
                System.out.println("接收消息===>"+textMessage.getText());
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    }
});
System.in.read();
//7 关闭资源
messageConsumer.close();
session.close();
connection.close();

3.3.4 消费者的三个情况

  • 1、先启动生产者，只启动1号消费者
    • 问：1号消费者能消费消息吗？
    • 答：可以
  • 2、先启动生产者，先启动1号消费者，再启动2号消费者
    • 问：2号消费者能消费消息吗？
    • 答：不可以，1号消费者全部消费完毕
  • 3、先启动2个消费者，再启动生产者生产6条消息
    • 问：消费信息的情况
    • 答：2个消费者平均分

3.4 Topic-测试代码实例

3.4.1 注意点

  • 1、生产者将消息发布到topic上，每个消息可以有多个消费者，属于1：N的关系。
  • 2、生产者和消费者之间有时间上的相关性，订阅某个主题的消费者只能消费自它订阅之后发布的消息。
  • 3、生产者生产时，topic不保存消息它是无状态的不落地，如果没有订阅，那就是一条废消息，所以一般先启动消费者再启动生产者。

3.4.2 生产者编码

只需要修改以下部分
```java
//4 创建目的地
Topic topic = session.createTopic(MY_TOPIC_NAME);

3.4.3 消费者编码

只需要修改以下部分

//4 创建目的地
Topic topic = session.createTopic(MY_TOPIC_NAME);

  • 控制台打印测试结果：
    image

  • web控制台：
    image

3.5 Queue和Topic模式的区别

image

四、JMS(Java Message Service)Java消息服务

JMS是javaEE的技术体系中之一。

4.1 组成结构

image

  • JMS Provider：MQ消息中间件服务器
  • JMS Producer：创建和发送JMS消息的应用
  • JMS Consumer：接收和处理JMS消息的应用
  • JMS Message：
    • 消息头：
      • JMSDestination：设置目的地：queue,topic
      • JMSDeliveryMode：持久和非持久
        • 持久：被传送一次仅仅一次，在JMS提供者出现故障，消息不会丢失，会在服务器恢复后再次传递
        • 非持久：最多传送一次,意味着服务器出现故障，消息回永久消失
      • JMSExpiration：在一定时间后过期，默认永久
      • JMSPriority：优先级，0-9个级别，0-4普通，5-9加急，默认4
      • JMSMessageID：唯一识别每个消息的标识，由MQ产生
        image
    • 消息体：
      • 封装具体消息
      • 发送和接收的消息体类型必须一致
      • 5种消息体格式：
        • TextMessage
        • MapMessage
        • BytesMessage
        • StreamMessage
        • ObjectMessage
    • 消息属性：
      • 需要除消息头字段以外的值
      • 识别/去重/重点标注等操作

4.2 JMS的可靠性

4.2.1 PERSISTENT：持久性

  • 参数设置：

    • 持久：messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);
    • 非持久：messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

  • 持久的Queue：

    • 默认是持久化

  • 持久的Topic：

    • 生产者代码实例：
    public class JmsProducerTopic_persistent {
    //url地址
    public static final String MY_ACTIVEMQ_URL = "tcp://192.168.10.100:61616";
    public static final String MY_TOPIC_NAME = "topic_dj";

    public static void main(String[] args) throws JMSException {
        //1 创建Connection连接工厂
        ActiveMQConnectionFactory acf = new ActiveMQConnectionFactory(MY_ACTIVEMQ_URL);
        //2 使用连接工厂创建连接并启动
        Connection connection = acf.createConnection();
        //3 创建会话Session,两个参数：第一个事务/第二个签收
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //4 创建目的地
        Topic topic = session.createTopic(MY_TOPIC_NAME);
        //5 创建消息生产者
        MessageProducer messageProducer = session.createProducer(topic);
        messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);
        connection.start();
        for (int i = 1; i <= 3; i++){
            //6 创建消息
            TextMessage message = session.createTextMessage("topic_dj消息msg--->" + i);
            //7 messageProducer将产生的消息发送到MQ队列中
            messageProducer.send(message);
        }
        //8 关闭资源
        messageProducer.close();
        session.close();
        connection.close();

        //测试程序完成
        System.out.println("*****程序发布成功");
        }
}
    • 消费者代码实例：
    public class JmsConsumerTopic_persistent {

    //url地址
    public static final String MY_ACTIVEMQ_URL = "tcp://192.168.10.100:61616";
    public static final String MY_TOPIC_NAME = "topic_dj";

    public static void main(String[] args) throws JMSException, IOException {

        System.out.println("我是1号topic消费者");

        //1 创建Connection连接工厂
        ActiveMQConnectionFactory acf = new ActiveMQConnectionFactory(MY_ACTIVEMQ_URL);
        //2 使用连接工厂创建连接并启动
        Connection connection = acf.createConnection();
        connection.setClientID("1号消费者");
        //3 创建会话Session,两个参数：第一个事务/第二个签收
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //4 创建目的地
        Topic topic = session.createTopic(MY_TOPIC_NAME);
        TopicSubscriber topicSubscriber = session.createDurableSubscriber(topic,"remark...");
        //5
        connection.start();
        Message message = topicSubscriber.receive();
        while (null != message){
            TextMessage textMessage = (TextMessage) message;
            System.out.println("****收到持久化topic："+textMessage.getText());
            message = topicSubscriber.receive(1000L);
        }

        session.close();
        connection.close();
    }
}
    • 启动前web控制台：

    image

    • 启动后web控制台：

      • 接收时间不过期

      image

      • 接收时间在一定时间过期

      image

      • 最后效果：

      image

4.2.2 事务：

  • 事务需要实现的代码实例：
//3 创建会话Session,两个参数：第一个事务/第二个签收
Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
......
session.commit();

4.2.3 签收：

  • 非事务：
    • 自动签收(默认)：Session.AUTO_ACKNOWLEDGE
    • 手动签收：Session.CLIENT_ACKNOWLEDGE
      • 客户端调用acknowledge方法手动签收
      try {
    System.out.println("接收消息===>"+textMessage.getText());
    message.acknowledge();
} catch (JMSException e) {
    e.printStackTrace();
}
    • 允许重复消息：Session.DUPS_OK_ACKNOWLEDGE
  • 事务：
    • 生产事务开启，只有commit后才能将全部消息变为已消费
    • 消息生产者
    • 消息消费者
  • 事务和签收的关系：
    • 在事务会话中，当一个事务被成功提交则消息会被自动签收。如果事务回滚，则消息会被再次传送
    • 非事务性会话中，消息何时被确认取决于创建会话时的应答模式

4.3 JMS的点对点总结

image

基于队列的，生产者发送消息到队列，消费者从队列接收消息，队列的存在使得消息的异步传输称为可能。类似于发短信

  • 如果Session关闭时，有部分消息已被接收但还没有被签收(acknowledge)，那当消费者下次连接到相同的队列时，这些消息还会被再次接收。
  • 队列可以长久的保存消息直到消费者收到消息，消费者不需要因为担心消息丢失而和队列时刻保持激活的来连接状态，充分体现了异步传输模式的优势

4.4 JMS的发布订阅总结

image

JMS Pub/Sub模型定义了如何向一个内容节点发布和订阅消息，这些节点被称为topic。类似于微信公众号

  • 主题可以被是消息的传输中介，发布者发布消息到主题，订阅者从主题订阅消息。
  • 主题使得消息订阅者和消息发布者保持互相独立，不需要接触即可保证消息的传送。
  • 非持久订阅下，不能恢复或者重新派送一个未签收的消息。
  • 持久订阅能恢复或重新派送未签收的消息。

五、ActiveMQ的Broker

5.1 概念

  • 相当于一个ActiveMQ服务器实例
  • Broker实现了用代码的形式启动ActiveMQ将MQ嵌入Java代码中，以便随时启动。
  • 在使用时启动，可以节省资源，保证可靠性。

5.2 conf配置文件

  • 不同的配置文件模拟不同的实例
    ./activemq start xbean:file:/myactiveMQ/apache-activemq-5.16.2/conf/activemq02.xml

5.3 嵌入式Broker

5.3.1 pom.xml

  • json绑定
<!-- https://mvnrepository.com/artifact/com.fasterxml.jackson.core/jackson-databind -->
<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
    <version>2.12.4</version>
</dependency>

5.3.2 EmbedBroker

public class EmbedBroker {

    public static void main(String[] args) throws Exception {
        //ActiveMQ也支持在vm中通信基于嵌入式的broker
        BrokerService brokerService = new BrokerService();
        brokerService.setUseJmx(true);
        brokerService.addConnector("tcp://localhost:61616");
        brokerService.start();
        System.in.read();
    }
}

5.3.3 队列验证

image

【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/jsit-dj-it/p/15132836.html