ActiveMQ和JMS介绍和简单的使用

逃离我推掉我的手 2022-02-19 13:31 276阅读 0赞

# 1 介绍 #

## 1.1 ActiveMQ ##

ActiveMQ是Apache提供的开源的消息系统，纯Java实现，因此很好的支持J2EE提出的JMS规范。

## 1.2 JMS ##

### 1.2.1 概念 ###

JMS（Java Message Service，即Java消息服务）是一组Java应用程序接口和相应的语法，类似于JDBC，是一组与厂商无关的API，使Java程序能与不同的厂商消息组件进行通信。  
消息发送异步进行，无须等待接受者立即响应。

### 1.2.2 消息类型 ###

1.  简单文本（TextMessage）。
2.  可序列化对象（ObjectMessage）。
3.  键值对（MapMessage）。
4.  字节流（BytesMessage）。
5.  流（StreamMessage）。

## 1.3 两种消息发送接收模型 ##

### 1.3.1 Point-to-Point（P2P），点对点 ###

该模型基于queue（队列），消息生产者发送消息到队列，消息消费者从队列中接收消息，队列的存在使得消息的异步成为可能。  
**消息发送接收**：过程多个生产者和消费者都可以注册到同一个消息队列，当生产者发送一个消息后，只有其中一个消费者会接收到消息，而不是所有的消费者都会接收到消息。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3F3cXczMzMzMzMz_size_16_color_FFFFFF_t_70]  
**代码示例**

/** * 消息生产者 */
    public class Producer { 
        public static void main(String[] args) throws InterruptedException { 
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
                    ActiveMQConnectionFactory.DEFAULT_USER,
                    ActiveMQConnectionFactory.DEFAULT_PASSWORD,
                    MQProperties.IP
            );
            try { 
                //通过 connection创建连接
                Connection connection=connectionFactory.createConnection();
                connection.start();
                //通过连接创建 session
                Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
    
                Destination destination= session.createQueue(MQProperties.QUEUE_NAME);
                MessageProducer producer = session.createProducer(destination);
                producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
                for (int i = 0; i < 10; i++) { 
                    ObjectMessage message = session.createObjectMessage("QUEUE计数器：【"+i+"】");
                    producer.send(message);
                    Thread.sleep(1000);
                    session.commit();
                }
    
            } catch (JMSException e) { 
                e.printStackTrace();
            }
        }
    }

/** * 消息消费者 */
    public class Consumer { 
        public static void main(String[] args) { 
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
                    ActiveMQConnectionFactory.DEFAULT_USER,
                    ActiveMQConnectionFactory.DEFAULT_PASSWORD,
                    MQProperties.IP
            );
            Connection connection;
            try { 
                connection = connectionFactory.createConnection();
                connection.start();
                Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
                Destination destination = session.createQueue(MQProperties.QUEUE_NAME);
                MessageConsumer consumer = session.createConsumer(destination);
                while (true) { 
                    ObjectMessage message = (ObjectMessage) consumer.receive(10000);
                    if (null != message) { 
                        System.out.println(message.getObject().toString());
                    }
                }
            } catch (JMSException e) { 
                e.printStackTrace();
            }
        }
    }

### 1.3.2 Pub/Sub（Publish/Subscribe），发布订阅 ###

发布/订阅模式定义了如何向一个内容节点发布和订阅消息，这个内容节点成为topic（主题）。主题可以认为是消息传递的中介，消息发布者将消息发布到某个主题，消息订阅者从主题中订阅消息。主题使发布者和订阅者相互保持独立，不需要接触即可保证消息的传递，该模式在一对多广播时采用。  
**消息发送接收**：消息发布者将消息投递给topic，消息订阅者需要在相应的topic进行注册，以便接受相应的topic消息。与点对点消息传递模型不同的是，消息发布者的消息将被自动发送给所有订阅该消息的订阅者。当消息订阅者由于某种原因与消息发布者断开连接时，这个时间段内的消息将会丢失，除非将消息订阅者模式设置为持久订阅（durable subscription），这时消息发布者会为消息订阅者保留这段时间内所生产的消息。当订阅者重新连接消息发布者时，订阅者仍然可以获取这部分消息。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3F3cXczMzMzMzMz_size_16_color_FFFFFF_t_70 1]  
**代码示例**

/** * 消息发布者 */
    public class Publisher { 
        public static void main(String[] args) throws InterruptedException { 
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
                    ActiveMQConnectionFactory.DEFAULT_USER,
                    ActiveMQConnectionFactory.DEFAULT_PASSWORD,
                    MQProperties.IP
            );
            try { 
                //通过 connection创建连接
                Connection connection=connectionFactory.createConnection();
                connection.start();
                //通过连接创建 session
                Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
                Topic topic = session.createTopic(MQProperties.TOPIC_NAME);
                MessageProducer producer = session.createProducer(topic);
                producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
                for (int i = 0; i < 10; i++) { 
                    TextMessage message = session.createTextMessage();
                    message.setText("TOPIC计数器：【"+i+"】");
                    producer.send(message);
                    Thread.sleep(1000);
                }
    
            } catch (JMSException e) { 
                e.printStackTrace();
            }
        }
    }

/** * 消息订阅者 */
    public class Subscriber { 
        public static void main(String[] args) { 
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
                    ActiveMQConnectionFactory.DEFAULT_USER,
                    ActiveMQConnectionFactory.DEFAULT_PASSWORD,
                    MQProperties.IP
            );
            Connection connection;
            try { 
                connection = connectionFactory.createConnection();
                connection.start();
                Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
                Topic topic = session.createTopic(MQProperties.TOPIC_NAME);
                MessageConsumer consumer = session.createConsumer(topic);
                consumer.setMessageListener(new MessageListener() { 
                    public void onMessage(Message message) { 
                        TextMessage tm = (TextMessage) message;
                        try { 
                            System.out.println(tm.getText());
                        } catch (JMSException e) { 
                            e.printStackTrace();
                        }
                    }
                });
            } catch (JMSException e) { 
                e.printStackTrace();
            }
        }
    }

*本文整理自陈康贤的《大型分布式网站架构》一书，特此感谢，如有侵权请联系本人删除*

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3F3cXczMzMzMzMz_size_16_color_FFFFFF_t_70]: /images/20220219/29203147db2e4992b9eeea3646b03586.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3F3cXczMzMzMzMz_size_16_color_FFFFFF_t_70 1]: /images/20220219/a0cfcf627fb04c5a8db4211031e69f0d.png